Wala shell

Advanced PHP File Manager

Server IP: 127.0.0.1 | PHP Version: 8.3.30 | Domains Hosted: 54

Name	Size	Last Modified	Actions
📁 ..	-	2026-03-21 15:45
📄 __init__.py	0 B	2026-03-10 23:36
📁 __pycache__	-	2026-03-21 15:45
📄 config.py # -- coding: utf-8 -- # # Copyright (C) 2010-2016 Red Hat, Inc. # # Authors: # Thomas Woerner <twoerner@redhat.com> # # This program is free software; you can redistribute it and/or modify # it under the terms of the GNU General Public License as published by # the Free Software Foundation; either version 2 of the License, or # (at your option) any later version. # # This program is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU General Public License for more details. # # You should have received a copy of the GNU General Public License # along with this program. If not, see <http://www.gnu.org/licenses/>. import os import dbus import dbus.service from firewall import config from firewall.core.base import DEFAULT_ZONE_TARGET from firewall.core.watcher import Watcher from firewall.core.logger import log from firewall.server.dbus import DbusServiceObject from firewall.server.decorators import handle_exceptions, \ dbus_handle_exceptions, dbus_service_method, \ dbus_service_method_deprecated, dbus_service_signal_deprecated, \ dbus_polkit_require_auth from firewall.server.config_icmptype import FirewallDConfigIcmpType from firewall.server.config_service import FirewallDConfigService from firewall.server.config_zone import FirewallDConfigZone from firewall.server.config_policy import FirewallDConfigPolicy from firewall.server.config_ipset import FirewallDConfigIPSet from firewall.server.config_helper import FirewallDConfigHelper from firewall.core.io.icmptype import IcmpType from firewall.core.io.ipset import IPSet from firewall.core.io.helper import Helper from firewall.core.io.lockdown_whitelist import LockdownWhitelist from firewall.core.io.direct import Direct from firewall.dbus_utils import dbus_to_python, \ command_of_sender, context_of_sender, uid_of_sender, user_of_uid, \ dbus_introspection_prepare_properties, \ dbus_introspection_add_properties, \ dbus_introspection_add_deprecated from firewall import errors from firewall.errors import FirewallError ############################################################################ # # class FirewallDConfig # ############################################################################ class FirewallDConfig(DbusServiceObject): """FirewallD main class""" persistent = True """ Make FirewallD persistent. """ default_polkit_auth_required = config.dbus.PK_ACTION_CONFIG """ Use config.dbus.PK_ACTION_INFO as a default """ @handle_exceptions def __init__(self, conf, args, kwargs): super(FirewallDConfig, self).__init__(args, **kwargs) self.config = conf self.busname = args[0] self.path = args[1] self._init_vars() self.watcher = Watcher(self.watch_updater, 5) self.watcher.add_watch_dir(config.FIREWALLD_IPSETS) self.watcher.add_watch_dir(config.ETC_FIREWALLD_IPSETS) self.watcher.add_watch_dir(config.FIREWALLD_ICMPTYPES) self.watcher.add_watch_dir(config.ETC_FIREWALLD_ICMPTYPES) self.watcher.add_watch_dir(config.FIREWALLD_HELPERS) self.watcher.add_watch_dir(config.ETC_FIREWALLD_HELPERS) self.watcher.add_watch_dir(config.FIREWALLD_SERVICES) self.watcher.add_watch_dir(config.ETC_FIREWALLD_SERVICES) self.watcher.add_watch_dir(config.FIREWALLD_ZONES) self.watcher.add_watch_dir(config.ETC_FIREWALLD_ZONES) self.watcher.add_watch_dir(config.FIREWALLD_POLICIES) self.watcher.add_watch_dir(config.ETC_FIREWALLD_POLICIES) # Add watches for combined zone directories if os.path.exists(config.ETC_FIREWALLD_ZONES): for filename in sorted(os.listdir(config.ETC_FIREWALLD_ZONES)): path = "%s/%s" % (config.ETC_FIREWALLD_ZONES, filename) if os.path.isdir(path): self.watcher.add_watch_dir(path) self.watcher.add_watch_file(config.LOCKDOWN_WHITELIST) self.watcher.add_watch_file(config.FIREWALLD_DIRECT) self.watcher.add_watch_file(config.FIREWALLD_CONF) dbus_introspection_prepare_properties(self, config.dbus.DBUS_INTERFACE_CONFIG, { "CleanupOnExit": "readwrite", "CleanupModulesOnExit": "readwrite", "IPv6_rpfilter": "readwrite", "IPv6_rpfilter2": "readwrite", "Lockdown": "readwrite", "MinimalMark": "readwrite", "IndividualCalls": "readwrite", "LogDenied": "readwrite", "AutomaticHelpers": "readwrite", "FirewallBackend": "readwrite", "FlushAllOnReload": "readwrite", "RFC3964_IPv4": "readwrite", "AllowZoneDrifting": "readwrite", "NftablesTableOwner": "readwrite", } ) @handle_exceptions def _init_vars(self): self.ipsets = [ ] self.ipset_idx = 0 self.icmptypes = [ ] self.icmptype_idx = 0 self.services = [ ] self.service_idx = 0 self.zones = [ ] self.zone_idx = 0 self.helpers = [ ] self.helper_idx = 0 self.policy_objects = [ ] self.policy_object_idx = 0 for ipset in self.config.get_ipsets(): self._addIPSet(self.config.get_ipset(ipset)) for icmptype in self.config.get_icmptypes(): self._addIcmpType(self.config.get_icmptype(icmptype)) for service in self.config.get_services(): self._addService(self.config.get_service(service)) for zone in self.config.get_zones(): self._addZone(self.config.get_zone(zone)) for helper in self.config.get_helpers(): self._addHelper(self.config.get_helper(helper)) for policy in self.config.get_policy_objects(): self._addPolicy(self.config.get_policy_object(policy)) @handle_exceptions def __del__(self): pass @handle_exceptions def reload(self): while len(self.ipsets) > 0: item = self.ipsets.pop() item.unregister() del item while len(self.icmptypes) > 0: item = self.icmptypes.pop() item.unregister() del item while len(self.services) > 0: item = self.services.pop() item.unregister() del item while len(self.zones) > 0: item = self.zones.pop() item.unregister() del item while len(self.helpers) > 0: item = self.helpers.pop() item.unregister() del item while len(self.policy_objects) > 0: item = self.policy_objects.pop() item.unregister() del item self._init_vars() @handle_exceptions def watch_updater(self, name): if name == config.FIREWALLD_CONF: old_props = self.GetAll(config.dbus.DBUS_INTERFACE_CONFIG) log.debug1("config: Reloading firewalld config file '%s'", config.FIREWALLD_CONF) try: self.config.update_firewalld_conf() except Exception as msg: log.error("Failed to load firewalld.conf file '%s': %s" % \ (name, msg)) return props = self.GetAll(config.dbus.DBUS_INTERFACE_CONFIG).copy() for key in list(props.keys()): if key in old_props and old_props[key] == props[key]: del props[key] if len(props) > 0: self.PropertiesChanged(config.dbus.DBUS_INTERFACE_CONFIG, props, []) return if (name.startswith(config.FIREWALLD_ICMPTYPES) or \ name.startswith(config.ETC_FIREWALLD_ICMPTYPES)) and \ name.endswith(".xml"): try: (what, obj) = self.config.update_icmptype_from_path(name) except Exception as msg: log.error("Failed to load icmptype file '%s': %s" % (name, msg)) return if what == "new": self._addIcmpType(obj) elif what == "remove": self.removeIcmpType(obj) elif what == "update": self._updateIcmpType(obj) elif (name.startswith(config.FIREWALLD_SERVICES) or \ name.startswith(config.ETC_FIREWALLD_SERVICES)) and \ name.endswith(".xml"): try: (what, obj) = self.config.update_service_from_path(name) except Exception as msg: log.error("Failed to load service file '%s': %s" % (name, msg)) return if what == "new": self._addService(obj) elif what == "remove": self.removeService(obj) elif what == "update": self._updateService(obj) elif name.startswith(config.FIREWALLD_ZONES) or \ name.startswith(config.ETC_FIREWALLD_ZONES): if name.endswith(".xml"): try: (what, obj) = self.config.update_zone_from_path(name) except Exception as msg: log.error("Failed to load zone file '%s': %s" % (name, msg)) return if what == "new": self._addZone(obj) elif what == "remove": self.removeZone(obj) elif what == "update": self._updateZone(obj) elif name.startswith(config.ETC_FIREWALLD_ZONES): # possible combined zone base directory _name = name.replace(config.ETC_FIREWALLD_ZONES, "").strip("/") if len(_name) < 1 or "/" in _name: # if there is a / in x, then it is a sub sub directory # ignore it return if os.path.isdir(name): if not self.watcher.has_watch(name): self.watcher.add_watch_dir(name) elif self.watcher.has_watch(name): self.watcher.remove_watch(name) elif (name.startswith(config.FIREWALLD_IPSETS) or \ name.startswith(config.ETC_FIREWALLD_IPSETS)) and \ name.endswith(".xml"): try: (what, obj) = self.config.update_ipset_from_path(name) except Exception as msg: log.error("Failed to load ipset file '%s': %s" % (name, msg)) return if what == "new": self._addIPSet(obj) elif what == "remove": self.removeIPSet(obj) elif what == "update": self._updateIPSet(obj) elif (name.startswith(config.FIREWALLD_HELPERS) or \ name.startswith(config.ETC_FIREWALLD_HELPERS)) and \ name.endswith(".xml"): try: (what, obj) = self.config.update_helper_from_path(name) except Exception as msg: log.error("Failed to load helper file '%s': %s" % (name, msg)) return if what == "new": self._addHelper(obj) elif what == "remove": self.removeHelper(obj) elif what == "update": self._updateHelper(obj) elif name == config.LOCKDOWN_WHITELIST: try: self.config.update_lockdown_whitelist() except Exception as msg: log.error("Failed to load lockdown whitelist file '%s': %s" % \ (name, msg)) return self.LockdownWhitelistUpdated() elif name == config.FIREWALLD_DIRECT: try: self.config.update_direct() except Exception as msg: log.error("Failed to load direct rules file '%s': %s" % (name, msg)) return self.Updated() elif (name.startswith(config.FIREWALLD_POLICIES) or \ name.startswith(config.ETC_FIREWALLD_POLICIES)) and \ name.endswith(".xml"): try: (what, obj) = self.config.update_policy_object_from_path(name) except Exception as msg: log.error("Failed to load policy file '%s': %s" % (name, msg)) return if what == "new": self._addPolicy(obj) elif what == "remove": self.removePolicy(obj) elif what == "update": self._updatePolicy(obj) @handle_exceptions def _addIcmpType(self, obj): # TODO: check for idx overflow config_icmptype = FirewallDConfigIcmpType( self, self.config, obj, self.icmptype_idx, self.busname, "%s/%d" % (config.dbus.DBUS_PATH_CONFIG_ICMPTYPE, self.icmptype_idx)) self.icmptypes.append(config_icmptype) self.icmptype_idx += 1 self.IcmpTypeAdded(obj.name) return config_icmptype @handle_exceptions def _updateIcmpType(self, obj): for icmptype in self.icmptypes: if icmptype.obj.name == obj.name and \ icmptype.obj.path == obj.path and \ icmptype.obj.filename == obj.filename: icmptype.obj = obj icmptype.Updated(obj.name) @handle_exceptions def removeIcmpType(self, obj): index = 7 # see IMPORT_EXPORT_STRUCTURE in class Zone(IO_Object) for zone in self.zones: settings = zone.getSettings() # if this IcmpType is used in a zone remove it from that zone first if obj.name in settings[index]: settings[index].remove(obj.name) zone.obj = self.config.set_zone_config(zone.obj, settings) zone.Updated(zone.obj.name) for policy in self.policy_objects: settings = policy.getSettings() # if this IcmpType is used in a policy remove it from that policy first if "icmp_blocks" in settings and obj.name in settings["icmp_blocks"]: settings["icmp_blocks"].remove(obj.name) policy.obj = self.config.set_policy_object_config_dict(policy.obj, settings) policy.Updated(policy.obj.name) for icmptype in self.icmptypes: if icmptype.obj == obj: icmptype.Removed(obj.name) icmptype.unregister() self.icmptypes.remove(icmptype) del icmptype @handle_exceptions def _addService(self, obj): # TODO: check for idx overflow config_service = FirewallDConfigService( self, self.config, obj, self.service_idx, self.busname, "%s/%d" % (config.dbus.DBUS_PATH_CONFIG_SERVICE, self.service_idx)) self.services.append(config_service) self.service_idx += 1 self.ServiceAdded(obj.name) return config_service @handle_exceptions def _updateService(self, obj): for service in self.services: if service.obj.name == obj.name and \ service.obj.path == obj.path and \ service.obj.filename == obj.filename: service.obj = obj service.Updated(obj.name) @handle_exceptions def removeService(self, obj): index = 5 # see IMPORT_EXPORT_STRUCTURE in class Zone(IO_Object) for zone in self.zones: settings = zone.getSettings() # if this Service is used in a zone remove it from that zone first if obj.name in settings[index]: settings[index].remove(obj.name) zone.obj = self.config.set_zone_config(zone.obj, settings) zone.Updated(zone.obj.name) for policy in self.policy_objects: settings = policy.getSettings() # if this Service is used in a policy remove it from that policy first if "services" in settings and obj.name in settings["services"]: settings["services"].remove(obj.name) policy.obj = self.config.set_policy_object_config_dict(policy.obj, settings) policy.Updated(policy.obj.name) for service in self.services: if service.obj == obj: service.Removed(obj.name) service.unregister() self.services.remove(service) del service @handle_exceptions def _addZone(self, obj): # TODO: check for idx overflow config_zone = FirewallDConfigZone( self, self.config, obj, self.zone_idx, self.busname, "%s/%d" % (config.dbus.DBUS_PATH_CONFIG_ZONE, self.zone_idx)) self.zones.append(config_zone) self.zone_idx += 1 self.ZoneAdded(obj.name) return config_zone @handle_exceptions def _updateZone(self, obj): for zone in self.zones: if zone.obj.name == obj.name and zone.obj.path == obj.path and \ zone.obj.filename == obj.filename: zone.obj = obj zone.Updated(obj.name) @handle_exceptions def removeZone(self, obj): for zone in self.zones: if zone.obj == obj: zone.Removed(obj.name) zone.unregister() self.zones.remove(zone) del zone @handle_exceptions def _addPolicy(self, obj): # TODO: check for idx overflow config_policy = FirewallDConfigPolicy( self, self.config, obj, self.policy_object_idx, self.busname, "%s/%d" % (config.dbus.DBUS_PATH_CONFIG_POLICY, self.policy_object_idx)) self.policy_objects.append(config_policy) self.policy_object_idx += 1 self.PolicyAdded(obj.name) return config_policy @handle_exceptions def _updatePolicy(self, obj): for policy in self.policy_objects: if policy.obj.name == obj.name and policy.obj.path == obj.path and \ policy.obj.filename == obj.filename: policy.obj = obj policy.Updated(obj.name) @handle_exceptions def removePolicy(self, obj): for policy in self.policy_objects: if policy.obj == obj: policy.Removed(obj.name) policy.unregister() self.policy_objects.remove(policy) del policy @handle_exceptions def _addIPSet(self, obj): # TODO: check for idx overflow config_ipset = FirewallDConfigIPSet( self, self.config, obj, self.ipset_idx, self.busname, "%s/%d" % (config.dbus.DBUS_PATH_CONFIG_IPSET, self.ipset_idx)) self.ipsets.append(config_ipset) self.ipset_idx += 1 self.IPSetAdded(obj.name) return config_ipset @handle_exceptions def _updateIPSet(self, obj): for ipset in self.ipsets: if ipset.obj.name == obj.name and ipset.obj.path == obj.path and \ ipset.obj.filename == obj.filename: ipset.obj = obj ipset.Updated(obj.name) @handle_exceptions def removeIPSet(self, obj): for ipset in self.ipsets: if ipset.obj == obj: ipset.Removed(obj.name) ipset.unregister() self.ipsets.remove(ipset) del ipset # access check @handle_exceptions def _addHelper(self, obj): # TODO: check for idx overflow config_helper = FirewallDConfigHelper( self, self.config, obj, self.helper_idx, self.busname, "%s/%d" % (config.dbus.DBUS_PATH_CONFIG_HELPER, self.helper_idx)) self.helpers.append(config_helper) self.helper_idx += 1 self.HelperAdded(obj.name) return config_helper @handle_exceptions def _updateHelper(self, obj): for helper in self.helpers: if helper.obj.name == obj.name and helper.obj.path == obj.path and \ helper.obj.filename == obj.filename: helper.obj = obj helper.Updated(obj.name) @handle_exceptions def removeHelper(self, obj): for helper in self.helpers: if helper.obj == obj: helper.Removed(obj.name) helper.unregister() self.helpers.remove(helper) del helper # access check @dbus_handle_exceptions def accessCheck(self, sender): if self.config._fw._state == "FAILED": raise FirewallError(errors.RUNNING_BUT_FAILED, "Changing permanent configuration is not allowed while " "firewalld is in FAILED state. The permanent " "configuration must be fixed and then firewalld " "restarted. Try `firewall-offline-cmd --check-config`.") if self.config.lockdown_enabled(): if sender is None: log.error("Lockdown not possible, sender not set.") return bus = dbus.SystemBus() context = context_of_sender(bus, sender) if self.config.access_check("context", context): return uid = uid_of_sender(bus, sender) if self.config.access_check("uid", uid): return user = user_of_uid(uid) if self.config.access_check("user", user): return command = command_of_sender(bus, sender) if self.config.access_check("command", command): return raise FirewallError(errors.ACCESS_DENIED, "lockdown is enabled") # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # P R O P E R T I E S @dbus_handle_exceptions def _get_property(self, prop): if prop not in [ "DefaultZone", "MinimalMark", "CleanupOnExit", "CleanupModulesOnExit", "Lockdown", "IPv6_rpfilter", "IndividualCalls", "LogDenied", "AutomaticHelpers", "FirewallBackend", "FlushAllOnReload", "RFC3964_IPv4", "AllowZoneDrifting", "IPv6_rpfilter2", "NftablesTableOwner" ]: raise dbus.exceptions.DBusException( "org.freedesktop.DBus.Error.InvalidArgs: " "Property '%s' does not exist" % prop) value = self.config.get_firewalld_conf().get(prop) if prop == "DefaultZone": if value is None: value = config.FALLBACK_ZONE return dbus.String(value) elif prop == "MinimalMark": if value is None: value = config.FALLBACK_MINIMAL_MARK else: value = int(value) return dbus.Int32(value) elif prop == "CleanupOnExit": if value is None: value = "yes" if config.FALLBACK_CLEANUP_ON_EXIT else "no" return dbus.String(value) elif prop == "CleanupModulesOnExit": if value is None: value = "yes" if config.FALLBACK_CLEANUP_MODULES_ON_EXIT else "no" return dbus.String(value) elif prop == "Lockdown": if value is None: value = "yes" if config.FALLBACK_LOCKDOWN else "no" return dbus.String(value) elif prop == "IPv6_rpfilter": if value is None or value != "no": return dbus.String("yes") else: return dbus.String("no") elif prop == "IPv6_rpfilter2": if value is None: value = config.FALLBACK_IPV6_RPFILTER return dbus.String(value) elif prop == "IndividualCalls": if value is None: value = "yes" if config.FALLBACK_INDIVIDUAL_CALLS else "no" return dbus.String(value) elif prop == "LogDenied": if value is None: value = config.FALLBACK_LOG_DENIED return dbus.String(value) elif prop == "AutomaticHelpers": if value is None: value = config.FALLBACK_AUTOMATIC_HELPERS return dbus.String(value) elif prop == "FirewallBackend": if value is None: value = config.FALLBACK_FIREWALL_BACKEND return dbus.String(value) elif prop == "FlushAllOnReload": if value is None: value = "yes" if config.FALLBACK_FLUSH_ALL_ON_RELOAD else "no" return dbus.String(value) elif prop == "RFC3964_IPv4": if value is None: value = "yes" if config.FALLBACK_RFC3964_IPV4 else "no" return dbus.String(value) elif prop == "AllowZoneDrifting": if value is None: value = "yes" if config.FALLBACK_ALLOW_ZONE_DRIFTING else "no" return dbus.String(value) elif prop == "NftablesTableOwner": if value is None: value = "yes" if config.FALLBACK_NFTABLES_TABLE_OWNER else "no" return dbus.String(value) @dbus_handle_exceptions def _get_dbus_property(self, prop): if prop == "DefaultZone": return dbus.String(self._get_property(prop)) elif prop == "MinimalMark": return dbus.Int32(self._get_property(prop)) elif prop == "CleanupOnExit": return dbus.String(self._get_property(prop)) elif prop == "CleanupModulesOnExit": return dbus.String(self._get_property(prop)) elif prop == "Lockdown": return dbus.String(self._get_property(prop)) elif prop == "IPv6_rpfilter": return dbus.String(self._get_property(prop)) elif prop == "IPv6_rpfilter2": return dbus.String(self._get_property(prop)) elif prop == "IndividualCalls": return dbus.String(self._get_property(prop)) elif prop == "LogDenied": return dbus.String(self._get_property(prop)) elif prop == "AutomaticHelpers": return dbus.String(self._get_property(prop)) elif prop == "FirewallBackend": return dbus.String(self._get_property(prop)) elif prop == "FlushAllOnReload": return dbus.String(self._get_property(prop)) elif prop == "RFC3964_IPv4": return dbus.String(self._get_property(prop)) elif prop == "AllowZoneDrifting": return dbus.String(self._get_property(prop)) elif prop == "NftablesTableOwner": return dbus.String(self._get_property(prop)) else: raise dbus.exceptions.DBusException( "org.freedesktop.DBus.Error.InvalidArgs: " "Property '%s' does not exist" % prop) @dbus_service_method(dbus.PROPERTIES_IFACE, in_signature='ss', out_signature='v') @dbus_handle_exceptions def Get(self, interface_name, property_name, sender=None): # pylint: disable=W0613 # get a property interface_name = dbus_to_python(interface_name, str) property_name = dbus_to_python(property_name, str) log.debug1("config.Get('%s', '%s')", interface_name, property_name) if interface_name == config.dbus.DBUS_INTERFACE_CONFIG: return self._get_dbus_property(property_name) elif interface_name in [ config.dbus.DBUS_INTERFACE_CONFIG_DIRECT, config.dbus.DBUS_INTERFACE_CONFIG_POLICIES ]: raise dbus.exceptions.DBusException( "org.freedesktop.DBus.Error.InvalidArgs: " "Property '%s' does not exist" % property_name) else: raise dbus.exceptions.DBusException( "org.freedesktop.DBus.Error.UnknownInterface: " "Interface '%s' does not exist" % interface_name) @dbus_service_method(dbus.PROPERTIES_IFACE, in_signature='s', out_signature='a{sv}') @dbus_handle_exceptions def GetAll(self, interface_name, sender=None): # pylint: disable=W0613 interface_name = dbus_to_python(interface_name, str) log.debug1("config.GetAll('%s')", interface_name) ret = { } if interface_name == config.dbus.DBUS_INTERFACE_CONFIG: for x in [ "DefaultZone", "MinimalMark", "CleanupOnExit", "CleanupModulesOnExit", "Lockdown", "IPv6_rpfilter", "IndividualCalls", "LogDenied", "AutomaticHelpers", "FirewallBackend", "FlushAllOnReload", "RFC3964_IPv4", "AllowZoneDrifting", "IPv6_rpfilter2", "NftablesTableOwner" ]: ret[x] = self._get_property(x) elif interface_name in [ config.dbus.DBUS_INTERFACE_CONFIG_DIRECT, config.dbus.DBUS_INTERFACE_CONFIG_POLICIES ]: pass else: raise dbus.exceptions.DBusException( "org.freedesktop.DBus.Error.UnknownInterface: " "Interface '%s' does not exist" % interface_name) return dbus.Dictionary(ret, signature="sv") @dbus_polkit_require_auth(config.dbus.PK_ACTION_CONFIG) @dbus_service_method(dbus.PROPERTIES_IFACE, in_signature='ssv') @dbus_handle_exceptions def Set(self, interface_name, property_name, new_value, sender=None): interface_name = dbus_to_python(interface_name, str) property_name = dbus_to_python(property_name, str) new_value = dbus_to_python(new_value) log.debug1("config.Set('%s', '%s', '%s')", interface_name, property_name, new_value) self.accessCheck(sender) if interface_name == config.dbus.DBUS_INTERFACE_CONFIG: if property_name in [ "CleanupOnExit", "CleanupModulesOnExit", "Lockdown", "IPv6_rpfilter", "IndividualCalls", "LogDenied", "FirewallBackend", "FlushAllOnReload", "RFC3964_IPv4", "IPv6_rpfilter2", "NftablesTableOwner" ]: if property_name in [ "CleanupOnExit", "CleanupModulesOnExit", "Lockdown", "IPv6_rpfilter", "IndividualCalls", "FlushAllOnReload", "RFC3964_IPv4", "NftablesTableOwner" ]: if new_value.lower() not in [ "yes", "no", "true", "false" ]: raise FirewallError(errors.INVALID_VALUE, "'%s' for %s" % \ (new_value, property_name)) elif property_name == "LogDenied": if new_value not in config.LOG_DENIED_VALUES: raise FirewallError(errors.INVALID_VALUE, "'%s' for %s" % \ (new_value, property_name)) elif property_name == "FirewallBackend": if new_value not in config.FIREWALL_BACKEND_VALUES: raise FirewallError(errors.INVALID_VALUE, "'%s' for %s" % \ (new_value, property_name)) elif property_name == "IPv6_rpfilter2": if new_value not in config.IPV6_RPFILTER_VALUES: raise FirewallError( errors.INVALID_VALUE, "'%s' for %s" % (new_value, property_name), ) else: raise dbus.exceptions.DBusException( "org.freedesktop.DBus.Error.InvalidArgs: " "Property '%s' does not exist" % property_name) self.config.get_firewalld_conf().set(property_name, new_value) self.config.get_firewalld_conf().write() self.PropertiesChanged(interface_name, { property_name: new_value }, [ ]) elif property_name in ["MinimalMark", "AutomaticHelpers", "AllowZoneDrifting"]: # deprecated fields. Ignore setting them. pass else: raise dbus.exceptions.DBusException( "org.freedesktop.DBus.Error.InvalidArgs: " "Property '%s' does not exist" % property_name) elif interface_name in [ config.dbus.DBUS_INTERFACE_CONFIG_DIRECT, config.dbus.DBUS_INTERFACE_CONFIG_POLICIES ]: raise dbus.exceptions.DBusException( "org.freedesktop.DBus.Error.InvalidArgs: " "Property '%s' does not exist" % property_name) else: raise dbus.exceptions.DBusException( "org.freedesktop.DBus.Error.UnknownInterface: " "Interface '%s' does not exist" % interface_name) @dbus.service.signal(dbus.PROPERTIES_IFACE, signature='sa{sv}as') def PropertiesChanged(self, interface_name, changed_properties, invalidated_properties): interface_name = dbus_to_python(interface_name, str) changed_properties = dbus_to_python(changed_properties) invalidated_properties = dbus_to_python(invalidated_properties) log.debug1("config.PropertiesChanged('%s', '%s', '%s')", interface_name, changed_properties, invalidated_properties) @dbus_polkit_require_auth(config.dbus.PK_ACTION_INFO) @dbus_service_method(dbus.INTROSPECTABLE_IFACE, out_signature='s') @dbus_handle_exceptions def Introspect(self, sender=None): # pylint: disable=W0613 log.debug2("config.Introspect()") data = super(FirewallDConfig, self).Introspect(self.path, self.busname.get_bus()) data = dbus_introspection_add_properties( self, data, config.dbus.DBUS_INTERFACE_CONFIG) for interface in [config.dbus.DBUS_INTERFACE_CONFIG_DIRECT]: data = dbus_introspection_add_deprecated( self, data, interface, dbus_service_method_deprecated().deprecated, dbus_service_signal_deprecated().deprecated) return data # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # policies @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_POLICIES, out_signature=LockdownWhitelist.DBUS_SIGNATURE) @dbus_handle_exceptions def getLockdownWhitelist(self, sender=None): # pylint: disable=W0613 log.debug1("config.policies.getLockdownWhitelist()") return self.config.get_policies().lockdown_whitelist.export_config() @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_POLICIES, in_signature=LockdownWhitelist.DBUS_SIGNATURE) @dbus_handle_exceptions def setLockdownWhitelist(self, settings, sender=None): # pylint: disable=W0613 log.debug1("config.policies.setLockdownWhitelist(...)") settings = dbus_to_python(settings) self.config.get_policies().lockdown_whitelist.import_config(settings, {}) self.config.get_policies().lockdown_whitelist.write() self.LockdownWhitelistUpdated() @dbus.service.signal(config.dbus.DBUS_INTERFACE_CONFIG_POLICIES) @dbus_handle_exceptions def LockdownWhitelistUpdated(self): log.debug1("config.policies.LockdownWhitelistUpdated()") # command @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_POLICIES, in_signature='s') @dbus_handle_exceptions def addLockdownWhitelistCommand(self, command, sender=None): command = dbus_to_python(command) log.debug1("config.policies.addLockdownWhitelistCommand('%s')", command) self.accessCheck(sender) settings = list(self.getLockdownWhitelist()) if command in settings[0]: raise FirewallError(errors.ALREADY_ENABLED, command) settings[0].append(command) self.setLockdownWhitelist(settings) @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_POLICIES, in_signature='s') @dbus_handle_exceptions def removeLockdownWhitelistCommand(self, command, sender=None): command = dbus_to_python(command) log.debug1("config.policies.removeLockdownWhitelistCommand('%s')", command) self.accessCheck(sender) settings = list(self.getLockdownWhitelist()) if command not in settings[0]: raise FirewallError(errors.NOT_ENABLED, command) settings[0].remove(command) self.setLockdownWhitelist(settings) @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_POLICIES, in_signature='s', out_signature='b') @dbus_handle_exceptions def queryLockdownWhitelistCommand(self, command, sender=None): # pylint: disable=W0613 command = dbus_to_python(command) log.debug1("config.policies.queryLockdownWhitelistCommand('%s')", command) return command in self.getLockdownWhitelist()[0] @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_POLICIES, out_signature='as') @dbus_handle_exceptions def getLockdownWhitelistCommands(self, sender=None): # pylint: disable=W0613 log.debug1("config.policies.getLockdownWhitelistCommands()") return self.getLockdownWhitelist()[0] # context @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_POLICIES, in_signature='s') @dbus_handle_exceptions def addLockdownWhitelistContext(self, context, sender=None): context = dbus_to_python(context) log.debug1("config.policies.addLockdownWhitelistContext('%s')", context) self.accessCheck(sender) settings = list(self.getLockdownWhitelist()) if context in settings[1]: raise FirewallError(errors.ALREADY_ENABLED, context) settings[1].append(context) self.setLockdownWhitelist(settings) @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_POLICIES, in_signature='s') @dbus_handle_exceptions def removeLockdownWhitelistContext(self, context, sender=None): context = dbus_to_python(context) log.debug1("config.policies.removeLockdownWhitelistContext('%s')", context) self.accessCheck(sender) settings = list(self.getLockdownWhitelist()) if context not in settings[1]: raise FirewallError(errors.NOT_ENABLED, context) settings[1].remove(context) self.setLockdownWhitelist(settings) @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_POLICIES, in_signature='s', out_signature='b') @dbus_handle_exceptions def queryLockdownWhitelistContext(self, context, sender=None): # pylint: disable=W0613 context = dbus_to_python(context) log.debug1("config.policies.queryLockdownWhitelistContext('%s')", context) return context in self.getLockdownWhitelist()[1] @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_POLICIES, out_signature='as') @dbus_handle_exceptions def getLockdownWhitelistContexts(self, sender=None): # pylint: disable=W0613 log.debug1("config.policies.getLockdownWhitelistContexts()") return self.getLockdownWhitelist()[1] # user @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_POLICIES, in_signature='s') @dbus_handle_exceptions def addLockdownWhitelistUser(self, user, sender=None): user = dbus_to_python(user) log.debug1("config.policies.addLockdownWhitelistUser('%s')", user) self.accessCheck(sender) settings = list(self.getLockdownWhitelist()) if user in settings[2]: raise FirewallError(errors.ALREADY_ENABLED, user) settings[2].append(user) self.setLockdownWhitelist(settings) @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_POLICIES, in_signature='s') @dbus_handle_exceptions def removeLockdownWhitelistUser(self, user, sender=None): user = dbus_to_python(user) log.debug1("config.policies.removeLockdownWhitelistUser('%s')", user) self.accessCheck(sender) settings = list(self.getLockdownWhitelist()) if user not in settings[2]: raise FirewallError(errors.NOT_ENABLED, user) settings[2].remove(user) self.setLockdownWhitelist(settings) @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_POLICIES, in_signature='s', out_signature='b') @dbus_handle_exceptions def queryLockdownWhitelistUser(self, user, sender=None): # pylint: disable=W0613 user = dbus_to_python(user) log.debug1("config.policies.queryLockdownWhitelistUser('%s')", user) return user in self.getLockdownWhitelist()[2] @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_POLICIES, out_signature='as') @dbus_handle_exceptions def getLockdownWhitelistUsers(self, sender=None): # pylint: disable=W0613 log.debug1("config.policies.getLockdownWhitelistUsers()") return self.getLockdownWhitelist()[2] # uid @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_POLICIES, in_signature='i') @dbus_handle_exceptions def addLockdownWhitelistUid(self, uid, sender=None): uid = dbus_to_python(uid) log.debug1("config.policies.addLockdownWhitelistUid(%d)", uid) self.accessCheck(sender) settings = list(self.getLockdownWhitelist()) if uid in settings[3]: raise FirewallError(errors.ALREADY_ENABLED, uid) settings[3].append(uid) self.setLockdownWhitelist(settings) @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_POLICIES, in_signature='i') @dbus_handle_exceptions def removeLockdownWhitelistUid(self, uid, sender=None): uid = dbus_to_python(uid) log.debug1("config.policies.removeLockdownWhitelistUid(%d)", uid) self.accessCheck(sender) settings = list(self.getLockdownWhitelist()) if uid not in settings[3]: raise FirewallError(errors.NOT_ENABLED, uid) settings[3].remove(uid) self.setLockdownWhitelist(settings) @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_POLICIES, in_signature='i', out_signature='b') @dbus_handle_exceptions def queryLockdownWhitelistUid(self, uid, sender=None): # pylint: disable=W0613 uid = dbus_to_python(uid) log.debug1("config.policies.queryLockdownWhitelistUid(%d)", uid) return uid in self.getLockdownWhitelist()[3] @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_POLICIES, out_signature='ai') @dbus_handle_exceptions def getLockdownWhitelistUids(self, sender=None): # pylint: disable=W0613 log.debug1("config.policies.getLockdownWhitelistUids()") return self.getLockdownWhitelist()[3] # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # I P S E T S @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG, out_signature='ao') @dbus_handle_exceptions def listIPSets(self, sender=None): # pylint: disable=W0613 """list ipsets objects paths """ log.debug1("config.listIPSets()") return self.ipsets @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG, out_signature='as') @dbus_handle_exceptions def getIPSetNames(self, sender=None): # pylint: disable=W0613 """get ipset names """ log.debug1("config.getIPSetNames()") ipsets = [ ] for obj in self.ipsets: ipsets.append(obj.obj.name) return sorted(ipsets) @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG, in_signature='s', out_signature='o') @dbus_handle_exceptions def getIPSetByName(self, ipset, sender=None): # pylint: disable=W0613 """object path of ipset with given name """ ipset = dbus_to_python(ipset, str) log.debug1("config.getIPSetByName('%s')", ipset) for obj in self.ipsets: if obj.obj.name == ipset: return obj raise FirewallError(errors.INVALID_IPSET, ipset) @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG, in_signature='s'+IPSet.DBUS_SIGNATURE, out_signature='o') @dbus_handle_exceptions def addIPSet(self, ipset, settings, sender=None): """add ipset with given name and settings """ ipset = dbus_to_python(ipset, str) settings = dbus_to_python(settings) log.debug1("config.addIPSet('%s')", ipset) self.accessCheck(sender) obj = self.config.new_ipset(ipset, settings) config_ipset = self._addIPSet(obj) return config_ipset @dbus.service.signal(config.dbus.DBUS_INTERFACE_CONFIG, signature='s') @dbus_handle_exceptions def IPSetAdded(self, ipset): ipset = dbus_to_python(ipset, str) log.debug1("config.IPSetAdded('%s')" % (ipset)) # I C M P T Y P E S @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG, out_signature='ao') @dbus_handle_exceptions def listIcmpTypes(self, sender=None): # pylint: disable=W0613 """list icmptypes objects paths """ log.debug1("config.listIcmpTypes()") return self.icmptypes @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG, out_signature='as') @dbus_handle_exceptions def getIcmpTypeNames(self, sender=None): # pylint: disable=W0613 """get icmptype names """ log.debug1("config.getIcmpTypeNames()") icmptypes = [ ] for obj in self.icmptypes: icmptypes.append(obj.obj.name) return sorted(icmptypes) @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG, in_signature='s', out_signature='o') @dbus_handle_exceptions def getIcmpTypeByName(self, icmptype, sender=None): # pylint: disable=W0613 """object path of icmptype with given name """ icmptype = dbus_to_python(icmptype, str) log.debug1("config.getIcmpTypeByName('%s')", icmptype) for obj in self.icmptypes: if obj.obj.name == icmptype: return obj raise FirewallError(errors.INVALID_ICMPTYPE, icmptype) @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG, in_signature='s'+IcmpType.DBUS_SIGNATURE, out_signature='o') @dbus_handle_exceptions def addIcmpType(self, icmptype, settings, sender=None): """add icmptype with given name and settings """ icmptype = dbus_to_python(icmptype, str) settings = dbus_to_python(settings) log.debug1("config.addIcmpType('%s')", icmptype) self.accessCheck(sender) obj = self.config.new_icmptype(icmptype, settings) config_icmptype = self._addIcmpType(obj) return config_icmptype @dbus.service.signal(config.dbus.DBUS_INTERFACE_CONFIG, signature='s') @dbus_handle_exceptions def IcmpTypeAdded(self, icmptype): log.debug1("config.IcmpTypeAdded('%s')" % (icmptype)) # S E R V I C E S @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG, out_signature='ao') @dbus_handle_exceptions def listServices(self, sender=None): # pylint: disable=W0613 """list services objects paths """ log.debug1("config.listServices()") return self.services @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG, out_signature='as') @dbus_handle_exceptions def getServiceNames(self, sender=None): # pylint: disable=W0613 """get service names """ log.debug1("config.getServiceNames()") services = [ ] for obj in self.services: services.append(obj.obj.name) return sorted(services) @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG, in_signature='s', out_signature='o') @dbus_handle_exceptions def getServiceByName(self, service, sender=None): # pylint: disable=W0613 """object path of service with given name """ service = dbus_to_python(service, str) log.debug1("config.getServiceByName('%s')", service) for obj in self.services: if obj.obj.name == service: return obj raise FirewallError(errors.INVALID_SERVICE, service) @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG, in_signature='s(sssa(ss)asa{ss}asa(ss))', out_signature='o') @dbus_handle_exceptions def addService(self, service, settings, sender=None): """add service with given name and settings """ service = dbus_to_python(service, str) settings = dbus_to_python(settings) log.debug1("config.addService('%s')", service) self.accessCheck(sender) obj = self.config.new_service(service, settings) config_service = self._addService(obj) return config_service @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG, in_signature='sa{sv}', out_signature='o') @dbus_handle_exceptions def addService2(self, service, settings, sender=None): """add service with given name and settings """ service = dbus_to_python(service, str) settings = dbus_to_python(settings) log.debug1("config.addService2('%s')", service) self.accessCheck(sender) obj = self.config.new_service_dict(service, settings) config_service = self._addService(obj) return config_service @dbus.service.signal(config.dbus.DBUS_INTERFACE_CONFIG, signature='s') @dbus_handle_exceptions def ServiceAdded(self, service): log.debug1("config.ServiceAdded('%s')" % (service)) # Z O N E S @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG, out_signature='ao') @dbus_handle_exceptions def listZones(self, sender=None): # pylint: disable=W0613 """list zones objects paths """ log.debug1("config.listZones()") return self.zones @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG, out_signature='as') @dbus_handle_exceptions def getZoneNames(self, sender=None): # pylint: disable=W0613 """get zone names """ log.debug1("config.getZoneNames()") zones = [ ] for obj in self.zones: zones.append(obj.obj.name) return sorted(zones) @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG, in_signature='s', out_signature='o') @dbus_handle_exceptions def getZoneByName(self, zone, sender=None): # pylint: disable=W0613 """object path of zone with given name """ zone = dbus_to_python(zone, str) log.debug1("config.getZoneByName('%s')", zone) for obj in self.zones: if obj.obj.name == zone: return obj raise FirewallError(errors.INVALID_ZONE, zone) @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG, in_signature='s', out_signature='s') @dbus_handle_exceptions def getZoneOfInterface(self, iface, sender=None): # pylint: disable=W0613 """name of zone the given interface belongs to """ iface = dbus_to_python(iface, str) log.debug1("config.getZoneOfInterface('%s')", iface) ret = [] for obj in self.zones: if iface in obj.obj.interfaces: ret.append(obj.obj.name) if len(ret) > 1: # Even it shouldn't happen, it's actually possible that # the same interface is in several zone XML files return " ".join(ret) + \ " (ERROR: interface '%s' is in %s zone XML files, can be only in one)" % \ (iface, len(ret)) return ret[0] if ret else "" @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG, in_signature='s', out_signature='s') @dbus_handle_exceptions def getZoneOfSource(self, source, sender=None): # pylint: disable=W0613 """name of zone the given source belongs to """ source = dbus_to_python(source, str) log.debug1("config.getZoneOfSource('%s')", source) ret = [] for obj in self.zones: if source in obj.obj.sources: ret.append(obj.obj.name) if len(ret) > 1: # Even it shouldn't happen, it's actually possible that # the same source is in several zone XML files return " ".join(ret) + \ " (ERROR: source '%s' is in %s zone XML files, can be only in one)" % \ (source, len(ret)) return ret[0] if ret else "" @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG, in_signature="s(sssbsasa(ss)asba(ssss)asasasasa(ss)b)", out_signature='o') @dbus_handle_exceptions def addZone(self, zone, settings, sender=None): """add zone with given name and settings """ zone = dbus_to_python(zone, str) settings = dbus_to_python(settings) log.debug1("config.addZone('%s')", zone) self.accessCheck(sender) if settings[4] == "default": # convert to list, fix target, convert back to tuple _settings = list(settings) _settings[4] = DEFAULT_ZONE_TARGET settings = tuple(_settings) obj = self.config.new_zone(zone, settings) config_zone = self._addZone(obj) return config_zone @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG, in_signature="sa{sv}", out_signature='o') @dbus_handle_exceptions def addZone2(self, zone, settings, sender=None): """add zone with given name and settings """ zone = dbus_to_python(zone, str) settings = dbus_to_python(settings) log.debug1("config.addZone('%s')", zone) self.accessCheck(sender) if "target" in settings and settings["target"] == "default": settings["target"] = DEFAULT_ZONE_TARGET obj = self.config.new_zone_dict(zone, settings) config_zone = self._addZone(obj) return config_zone @dbus.service.signal(config.dbus.DBUS_INTERFACE_CONFIG, signature='s') @dbus_handle_exceptions def ZoneAdded(self, zone): log.debug1("config.ZoneAdded('%s')" % (zone)) # policies @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG, out_signature='ao') @dbus_handle_exceptions def listPolicies(self, sender=None): """list policies objects paths """ log.debug1("config.listPolicies()") return self.policy_objects @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG, out_signature='as') @dbus_handle_exceptions def getPolicyNames(self, sender=None): """get policy names """ log.debug1("config.getPolicyNames()") policies = [ ] for obj in self.policy_objects: policies.append(obj.obj.name) return sorted(policies) @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG, in_signature='s', out_signature='o') @dbus_handle_exceptions def getPolicyByName(self, policy, sender=None): """object path of policy with given name """ policy = dbus_to_python(policy, str) log.debug1("config.getPolicyByName('%s')", policy) for obj in self.policy_objects: if obj.obj.name == policy: return obj raise FirewallError(errors.INVALID_POLICY, policy) @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG, in_signature="sa{sv}", out_signature='o') @dbus_handle_exceptions def addPolicy(self, policy, settings, sender=None): """add policy with given name and settings """ policy = dbus_to_python(policy, str) settings = dbus_to_python(settings) log.debug1("config.addPolicy('%s')", policy) self.accessCheck(sender) obj = self.config.new_policy_object_dict(policy, settings) config_policy = self._addPolicy(obj) return config_policy @dbus.service.signal(config.dbus.DBUS_INTERFACE_CONFIG, signature='s') @dbus_handle_exceptions def PolicyAdded(self, policy): log.debug1("config.PolicyAdded('%s')" % (policy)) # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # H E L P E R S @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG, out_signature='ao') @dbus_handle_exceptions def listHelpers(self, sender=None): # pylint: disable=W0613 """list helpers objects paths """ log.debug1("config.listHelpers()") return self.helpers @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG, out_signature='as') @dbus_handle_exceptions def getHelperNames(self, sender=None): # pylint: disable=W0613 """get helper names """ log.debug1("config.getHelperNames()") helpers = [ ] for obj in self.helpers: helpers.append(obj.obj.name) return sorted(helpers) @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG, in_signature='s', out_signature='o') @dbus_handle_exceptions def getHelperByName(self, helper, sender=None): # pylint: disable=W0613 """object path of helper with given name """ helper = dbus_to_python(helper, str) log.debug1("config.getHelperByName('%s')", helper) for obj in self.helpers: if obj.obj.name == helper: return obj raise FirewallError(errors.INVALID_HELPER, helper) @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG, in_signature='s'+Helper.DBUS_SIGNATURE, out_signature='o') @dbus_handle_exceptions def addHelper(self, helper, settings, sender=None): """add helper with given name and settings """ helper = dbus_to_python(helper, str) settings = dbus_to_python(settings) log.debug1("config.addHelper('%s')", helper) self.accessCheck(sender) obj = self.config.new_helper(helper, settings) config_helper = self._addHelper(obj) return config_helper @dbus.service.signal(config.dbus.DBUS_INTERFACE_CONFIG, signature='s') @dbus_handle_exceptions def HelperAdded(self, helper): helper = dbus_to_python(helper, str) log.debug1("config.HelperAdded('%s')" % (helper)) # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # DIRECT @dbus_service_method_deprecated(config.dbus.DBUS_INTERFACE_CONFIG_DIRECT) @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_DIRECT, out_signature=Direct.DBUS_SIGNATURE) @dbus_handle_exceptions def getSettings(self, sender=None): # pylint: disable=W0613 # returns list ipv, table, list of chains log.debug1("config.direct.getSettings()") return self.config.get_direct().export_config() @dbus_service_method_deprecated(config.dbus.DBUS_INTERFACE_CONFIG_DIRECT) @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_DIRECT, in_signature=Direct.DBUS_SIGNATURE) @dbus_handle_exceptions def update(self, settings, sender=None): # pylint: disable=W0613 # returns list ipv, table, list of chains log.debug1("config.direct.update()") settings = dbus_to_python(settings) self.config.get_direct().import_config(settings, {}) self.config.get_direct().write() self.Updated() @dbus_service_signal_deprecated(config.dbus.DBUS_INTERFACE_CONFIG_DIRECT) @dbus.service.signal(config.dbus.DBUS_INTERFACE_CONFIG_DIRECT) @dbus_handle_exceptions def Updated(self): log.debug1("config.direct.Updated()") # chain @dbus_service_method_deprecated(config.dbus.DBUS_INTERFACE_CONFIG_DIRECT) @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_DIRECT, in_signature='sss') @dbus_handle_exceptions def addChain(self, ipv, table, chain, sender=None): ipv = dbus_to_python(ipv) table = dbus_to_python(table) chain = dbus_to_python(chain) log.debug1("config.direct.addChain('%s', '%s', '%s')" % \ (ipv, table, chain)) self.accessCheck(sender) idx = tuple((ipv, table, chain)) settings = list(self.getSettings()) if idx in settings[0]: raise FirewallError(errors.ALREADY_ENABLED, "chain '%s' already is in '%s:%s'" % \ (chain, ipv, table)) settings[0].append(idx) self.update(settings) @dbus_service_method_deprecated(config.dbus.DBUS_INTERFACE_CONFIG_DIRECT) @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_DIRECT, in_signature='sss') @dbus_handle_exceptions def removeChain(self, ipv, table, chain, sender=None): ipv = dbus_to_python(ipv) table = dbus_to_python(table) chain = dbus_to_python(chain) log.debug1("config.direct.removeChain('%s', '%s', '%s')" % \ (ipv, table, chain)) self.accessCheck(sender) idx = tuple((ipv, table, chain)) settings = list(self.getSettings()) if idx not in settings[0]: raise FirewallError(errors.NOT_ENABLED, "chain '%s' is not in '%s:%s'" % (chain, ipv, table)) settings[0].remove(idx) self.update(settings) @dbus_service_method_deprecated(config.dbus.DBUS_INTERFACE_CONFIG_DIRECT) @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_DIRECT, in_signature='sss', out_signature='b') @dbus_handle_exceptions def queryChain(self, ipv, table, chain, sender=None): # pylint: disable=W0613 ipv = dbus_to_python(ipv) table = dbus_to_python(table) chain = dbus_to_python(chain) log.debug1("config.direct.queryChain('%s', '%s', '%s')" % \ (ipv, table, chain)) idx = tuple((ipv, table, chain)) return idx in self.getSettings()[0] @dbus_service_method_deprecated(config.dbus.DBUS_INTERFACE_CONFIG_DIRECT) @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_DIRECT, in_signature='ss', out_signature='as') @dbus_handle_exceptions def getChains(self, ipv, table, sender=None): # pylint: disable=W0613 ipv = dbus_to_python(ipv) table = dbus_to_python(table) log.debug1("config.direct.getChains('%s', '%s')" % (ipv, table)) ret = [ ] for idx in self.getSettings()[0]: if idx[0] == ipv and idx[1] == table: ret.append(idx[2]) return ret @dbus_service_method_deprecated(config.dbus.DBUS_INTERFACE_CONFIG_DIRECT) @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_DIRECT, in_signature='', out_signature='a(sss)') @dbus_handle_exceptions def getAllChains(self, sender=None): # pylint: disable=W0613 log.debug1("config.direct.getAllChains()") return self.getSettings()[0] # rule @dbus_service_method_deprecated(config.dbus.DBUS_INTERFACE_CONFIG_DIRECT) @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_DIRECT, in_signature='sssias') @dbus_handle_exceptions def addRule(self, ipv, table, chain, priority, args, sender=None): # pylint: disable=R0913 ipv = dbus_to_python(ipv) table = dbus_to_python(table) chain = dbus_to_python(chain) priority = dbus_to_python(priority) args = dbus_to_python(args) log.debug1("config.direct.addRule('%s', '%s', '%s', %d, '%s')" % \ (ipv, table, chain, priority, "','".join(args))) self.accessCheck(sender) idx = (ipv, table, chain, priority, args) settings = list(self.getSettings()) if idx in settings[1]: raise FirewallError(errors.ALREADY_ENABLED, "rule '%s' already is in '%s:%s:%s'" % \ (args, ipv, table, chain)) settings[1].append(idx) self.update(tuple(settings)) @dbus_service_method_deprecated(config.dbus.DBUS_INTERFACE_CONFIG_DIRECT) @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_DIRECT, in_signature='sssias') @dbus_handle_exceptions def removeRule(self, ipv, table, chain, priority, args, sender=None): # pylint: disable=R0913 ipv = dbus_to_python(ipv) table = dbus_to_python(table) chain = dbus_to_python(chain) priority = dbus_to_python(priority) args = dbus_to_python(args) log.debug1("config.direct.removeRule('%s', '%s', '%s', %d, '%s')" % \ (ipv, table, chain, priority, "','".join(args))) self.accessCheck(sender) idx = (ipv, table, chain, priority, args) settings = list(self.getSettings()) if idx not in settings[1]: raise FirewallError(errors.NOT_ENABLED, "rule '%s' is not in '%s:%s:%s'" % \ (args, ipv, table, chain)) settings[1].remove(idx) self.update(tuple(settings)) @dbus_service_method_deprecated(config.dbus.DBUS_INTERFACE_CONFIG_DIRECT) @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_DIRECT, in_signature='sssias', out_signature='b') @dbus_handle_exceptions def queryRule(self, ipv, table, chain, priority, args, sender=None): # pylint: disable=W0613,R0913 ipv = dbus_to_python(ipv) table = dbus_to_python(table) chain = dbus_to_python(chain) priority = dbus_to_python(priority) args = dbus_to_python(args) log.debug1("config.direct.queryRule('%s', '%s', '%s', %d, '%s')" % \ (ipv, table, chain, priority, "','".join(args))) idx = (ipv, table, chain, priority, args) return idx in self.getSettings()[1] @dbus_service_method_deprecated(config.dbus.DBUS_INTERFACE_CONFIG_DIRECT) @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_DIRECT, in_signature='sss') @dbus_handle_exceptions def removeRules(self, ipv, table, chain, sender=None): ipv = dbus_to_python(ipv) table = dbus_to_python(table) chain = dbus_to_python(chain) log.debug1("config.direct.removeRules('%s', '%s', '%s')" % \ (ipv, table, chain, )) self.accessCheck(sender) settings = list(self.getSettings()) for rule in settings[1][:]: if (ipv, table, chain) == (rule[0], rule[1], rule[2]): settings[1].remove(rule) self.update(tuple(settings)) @dbus_service_method_deprecated(config.dbus.DBUS_INTERFACE_CONFIG_DIRECT) @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_DIRECT, in_signature='sss', out_signature='a(ias)') @dbus_handle_exceptions def getRules(self, ipv, table, chain, sender=None): # pylint: disable=W0613 ipv = dbus_to_python(ipv) table = dbus_to_python(table) chain = dbus_to_python(chain) log.debug1("config.direct.getRules('%s', '%s', '%s')" % \ (ipv, table, chain)) ret = [ ] for idx in self.getSettings()[1]: if idx[0] == ipv and idx[1] == table and idx[2] == chain: ret.append((idx[3], idx[4])) return ret @dbus_service_method_deprecated(config.dbus.DBUS_INTERFACE_CONFIG_DIRECT) @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_DIRECT, in_signature='', out_signature='a(sssias)') @dbus_handle_exceptions def getAllRules(self, sender=None): # pylint: disable=W0613 log.debug1("config.direct.getAllRules()") return self.getSettings()[1] # passthrough @dbus_service_method_deprecated(config.dbus.DBUS_INTERFACE_CONFIG_DIRECT) @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_DIRECT, in_signature='sas') @dbus_handle_exceptions def addPassthrough(self, ipv, args, sender=None): ipv = dbus_to_python(ipv) args = dbus_to_python(args) log.debug1("config.direct.addPassthrough('%s', '%s')" % \ (ipv, "','".join(args))) self.accessCheck(sender) idx = (ipv, args) settings = list(self.getSettings()) if idx in settings[2]: raise FirewallError(errors.ALREADY_ENABLED, "passthrough '%s', '%s'" % (ipv, args)) settings[2].append(idx) self.update(settings) @dbus_service_method_deprecated(config.dbus.DBUS_INTERFACE_CONFIG_DIRECT) @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_DIRECT, in_signature='sas') @dbus_handle_exceptions def removePassthrough(self, ipv, args, sender=None): ipv = dbus_to_python(ipv) args = dbus_to_python(args) log.debug1("config.direct.removePassthrough('%s', '%s')" % \ (ipv, "','".join(args))) self.accessCheck(sender) idx = (ipv, args) settings = list(self.getSettings()) if idx not in settings[2]: raise FirewallError(errors.NOT_ENABLED, "passthrough '%s', '%s'" % (ipv, args)) settings[2].remove(idx) self.update(settings) @dbus_service_method_deprecated(config.dbus.DBUS_INTERFACE_CONFIG_DIRECT) @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_DIRECT, in_signature='sas', out_signature='b') @dbus_handle_exceptions def queryPassthrough(self, ipv, args, sender=None): # pylint: disable=W0613 ipv = dbus_to_python(ipv) args = dbus_to_python(args) log.debug1("config.direct.queryPassthrough('%s', '%s')" % \ (ipv, "','".join(args))) idx = (ipv, args) return idx in self.getSettings()[2] @dbus_service_method_deprecated(config.dbus.DBUS_INTERFACE_CONFIG_DIRECT) @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_DIRECT, in_signature='s', out_signature='aas') @dbus_handle_exceptions def getPassthroughs(self, ipv, sender=None): # pylint: disable=W0613 ipv = dbus_to_python(ipv) log.debug1("config.direct.getPassthroughs('%s')" % (ipv)) ret = [ ] for idx in self.getSettings()[2]: if idx[0] == ipv: ret.append(idx[1]) return ret @dbus_service_method_deprecated(config.dbus.DBUS_INTERFACE_CONFIG_DIRECT) @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_DIRECT, out_signature='a(sas)') @dbus_handle_exceptions def getAllPassthroughs(self, sender=None): # pylint: disable=W0613 log.debug1("config.direct.getAllPassthroughs()") return self.getSettings()[2]	71.11 KB	2026-03-10 23:36
📄 config_helper.py # -- coding: utf-8 -- # # Copyright (C) 2010-2016 Red Hat, Inc. # # Authors: # Thomas Woerner <twoerner@redhat.com> # # This program is free software; you can redistribute it and/or modify # it under the terms of the GNU General Public License as published by # the Free Software Foundation; either version 2 of the License, or # (at your option) any later version. # # This program is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU General Public License for more details. # # You should have received a copy of the GNU General Public License # along with this program. If not, see <http://www.gnu.org/licenses/>. import dbus import dbus.service from firewall import config from firewall.dbus_utils import dbus_to_python, \ dbus_introspection_prepare_properties, \ dbus_introspection_add_properties from firewall.core.io.helper import Helper from firewall.core.logger import log from firewall.server.dbus import DbusServiceObject from firewall.server.decorators import handle_exceptions, \ dbus_handle_exceptions, dbus_service_method, \ dbus_polkit_require_auth from firewall import errors from firewall.errors import FirewallError ############################################################################ # # class FirewallDConfig # ############################################################################ class FirewallDConfigHelper(DbusServiceObject): """FirewallD main class""" persistent = True """ Make FirewallD persistent. """ default_polkit_auth_required = config.dbus.PK_ACTION_CONFIG """ Use PK_ACTION_INFO as a default """ @handle_exceptions def __init__(self, parent, conf, helper, item_id, args, kwargs): super(FirewallDConfigHelper, self).__init__(args, **kwargs) self.parent = parent self.config = conf self.obj = helper self.item_id = item_id self.busname = args[0] self.path = args[1] self._log_prefix = "config.helper.%d" % self.item_id dbus_introspection_prepare_properties( self, config.dbus.DBUS_INTERFACE_CONFIG_HELPER) @dbus_handle_exceptions def __del__(self): pass @dbus_handle_exceptions def unregister(self): self.remove_from_connection() # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # P R O P E R T I E S @dbus_handle_exceptions def _get_property(self, property_name): if property_name == "name": return dbus.String(self.obj.name) elif property_name == "filename": return dbus.String(self.obj.filename) elif property_name == "path": return dbus.String(self.obj.path) elif property_name == "default": return dbus.Boolean(self.obj.default) elif property_name == "builtin": return dbus.Boolean(self.obj.builtin) else: raise dbus.exceptions.DBusException( "org.freedesktop.DBus.Error.InvalidArgs: " "Property '%s' does not exist" % property_name) @dbus_service_method(dbus.PROPERTIES_IFACE, in_signature='ss', out_signature='v') @dbus_handle_exceptions def Get(self, interface_name, property_name, sender=None): # pylint: disable=W0613 # get a property interface_name = dbus_to_python(interface_name, str) property_name = dbus_to_python(property_name, str) log.debug1("%s.Get('%s', '%s')", self._log_prefix, interface_name, property_name) if interface_name != config.dbus.DBUS_INTERFACE_CONFIG_HELPER: raise dbus.exceptions.DBusException( "org.freedesktop.DBus.Error.UnknownInterface: " "Interface '%s' does not exist" % interface_name) return self._get_property(property_name) @dbus_service_method(dbus.PROPERTIES_IFACE, in_signature='s', out_signature='a{sv}') @dbus_handle_exceptions def GetAll(self, interface_name, sender=None): # pylint: disable=W0613 interface_name = dbus_to_python(interface_name, str) log.debug1("%s.GetAll('%s')", self._log_prefix, interface_name) if interface_name != config.dbus.DBUS_INTERFACE_CONFIG_HELPER: raise dbus.exceptions.DBusException( "org.freedesktop.DBus.Error.UnknownInterface: " "Interface '%s' does not exist" % interface_name) ret = { } for x in [ "name", "filename", "path", "default", "builtin" ]: ret[x] = self._get_property(x) return dbus.Dictionary(ret, signature="sv") @dbus_polkit_require_auth(config.dbus.PK_ACTION_CONFIG) @dbus_service_method(dbus.PROPERTIES_IFACE, in_signature='ssv') @dbus_handle_exceptions def Set(self, interface_name, property_name, new_value, sender=None): interface_name = dbus_to_python(interface_name, str) property_name = dbus_to_python(property_name, str) new_value = dbus_to_python(new_value) log.debug1("%s.Set('%s', '%s', '%s')", self._log_prefix, interface_name, property_name, new_value) self.parent.accessCheck(sender) if interface_name != config.dbus.DBUS_INTERFACE_CONFIG_HELPER: raise dbus.exceptions.DBusException( "org.freedesktop.DBus.Error.UnknownInterface: " "Interface '%s' does not exist" % interface_name) raise dbus.exceptions.DBusException( "org.freedesktop.DBus.Error.PropertyReadOnly: " "Property '%s' is read-only" % property_name) @dbus.service.signal(dbus.PROPERTIES_IFACE, signature='sa{sv}as') def PropertiesChanged(self, interface_name, changed_properties, invalidated_properties): interface_name = dbus_to_python(interface_name, str) changed_properties = dbus_to_python(changed_properties) invalidated_properties = dbus_to_python(invalidated_properties) log.debug1("%s.PropertiesChanged('%s', '%s', '%s')", self._log_prefix, interface_name, changed_properties, invalidated_properties) @dbus_polkit_require_auth(config.dbus.PK_ACTION_INFO) @dbus_service_method(dbus.INTROSPECTABLE_IFACE, out_signature='s') @dbus_handle_exceptions def Introspect(self, sender=None): # pylint: disable=W0613 log.debug2("%s.Introspect()", self._log_prefix) data = super(FirewallDConfigHelper, self).Introspect( self.path, self.busname.get_bus()) return dbus_introspection_add_properties( self, data, config.dbus.DBUS_INTERFACE_CONFIG_HELPER) # S E T T I N G S @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_HELPER, out_signature=Helper.DBUS_SIGNATURE) @dbus_handle_exceptions def getSettings(self, sender=None): # pylint: disable=W0613 """get settings for helper """ log.debug1("%s.getSettings()", self._log_prefix) return self.config.get_helper_config(self.obj) @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_HELPER, in_signature=Helper.DBUS_SIGNATURE) @dbus_handle_exceptions def update(self, settings, sender=None): """update settings for helper """ settings = dbus_to_python(settings) log.debug1("%s.update('...')", self._log_prefix) self.parent.accessCheck(sender) self.obj = self.config.set_helper_config(self.obj, settings) self.Updated(self.obj.name) @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_HELPER) @dbus_handle_exceptions def loadDefaults(self, sender=None): """load default settings for builtin helper """ log.debug1("%s.loadDefaults()", self._log_prefix) self.parent.accessCheck(sender) self.obj = self.config.load_helper_defaults(self.obj) self.Updated(self.obj.name) @dbus.service.signal(config.dbus.DBUS_INTERFACE_CONFIG_HELPER, signature='s') @dbus_handle_exceptions def Updated(self, name): log.debug1("%s.Updated('%s')" % (self._log_prefix, name)) # R E M O V E @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_HELPER) @dbus_handle_exceptions def remove(self, sender=None): """remove helper """ log.debug1("%s.removeHelper()", self._log_prefix) self.parent.accessCheck(sender) self.config.remove_helper(self.obj) self.parent.removeHelper(self.obj) @dbus.service.signal(config.dbus.DBUS_INTERFACE_CONFIG_HELPER, signature='s') @dbus_handle_exceptions def Removed(self, name): log.debug1("%s.Removed('%s')" % (self._log_prefix, name)) # R E N A M E @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_HELPER, in_signature='s') @dbus_handle_exceptions def rename(self, name, sender=None): """rename helper """ name = dbus_to_python(name, str) log.debug1("%s.rename('%s')", self._log_prefix, name) self.parent.accessCheck(sender) self.obj = self.config.rename_helper(self.obj, name) self.Renamed(name) @dbus.service.signal(config.dbus.DBUS_INTERFACE_CONFIG_HELPER, signature='s') @dbus_handle_exceptions def Renamed(self, name): log.debug1("%s.Renamed('%s')" % (self._log_prefix, name)) # version @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_HELPER, out_signature='s') @dbus_handle_exceptions def getVersion(self, sender=None): # pylint: disable=W0613 log.debug1("%s.getVersion()", self._log_prefix) return self.getSettings()[0] @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_HELPER, in_signature='s') @dbus_handle_exceptions def setVersion(self, version, sender=None): version = dbus_to_python(version, str) log.debug1("%s.setVersion('%s')", self._log_prefix, version) self.parent.accessCheck(sender) settings = list(self.getSettings()) settings[0] = version self.update(settings) # short @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_HELPER, out_signature='s') @dbus_handle_exceptions def getShort(self, sender=None): # pylint: disable=W0613 log.debug1("%s.getShort()", self._log_prefix) return self.getSettings()[1] @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_HELPER, in_signature='s') @dbus_handle_exceptions def setShort(self, short, sender=None): short = dbus_to_python(short, str) log.debug1("%s.setShort('%s')", self._log_prefix, short) self.parent.accessCheck(sender) settings = list(self.getSettings()) settings[1] = short self.update(settings) # description @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_HELPER, out_signature='s') @dbus_handle_exceptions def getDescription(self, sender=None): # pylint: disable=W0613 log.debug1("%s.getDescription()", self._log_prefix) return self.getSettings()[2] @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_HELPER, in_signature='s') @dbus_handle_exceptions def setDescription(self, description, sender=None): description = dbus_to_python(description, str) log.debug1("%s.setDescription('%s')", self._log_prefix, description) self.parent.accessCheck(sender) settings = list(self.getSettings()) settings[2] = description self.update(settings) # family @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_HELPER, out_signature='s') @dbus_handle_exceptions def getFamily(self, sender=None): log.debug1("%s.getFamily()", self._log_prefix) self.parent.accessCheck(sender) settings = list(self.getSettings()) return settings[3] @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_HELPER, in_signature='s') @dbus_handle_exceptions def setFamily(self, ipv, sender=None): ipv = dbus_to_python(ipv, str) log.debug1("%s.setFamily('%s')", self._log_prefix, ipv) self.parent.accessCheck(sender) settings = list(self.getSettings()) if settings[3] == ipv: raise FirewallError(errors.ALREADY_ENABLED, "'%s'" % ipv) settings[3] = ipv self.update(settings) @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_HELPER, in_signature='s', out_signature='b') @dbus_handle_exceptions def queryFamily(self, ipv, sender=None): # pylint: disable=W0613 ipv = dbus_to_python(ipv, str) log.debug1("%s.queryFamily('%s')", self._log_prefix, ipv) settings = self.getSettings() return (settings[3] == ipv) # module @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_HELPER, out_signature='s') @dbus_handle_exceptions def getModule(self, sender=None): log.debug1("%s.getModule()", self._log_prefix) self.parent.accessCheck(sender) settings = list(self.getSettings()) return settings[4] @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_HELPER, in_signature='s') @dbus_handle_exceptions def setModule(self, module, sender=None): module = dbus_to_python(module, str) log.debug1("%s.setModule('%s')", self._log_prefix, module) self.parent.accessCheck(sender) settings = list(self.getSettings()) if settings[4] == module: raise FirewallError(errors.ALREADY_ENABLED, "'%s'" % module) settings[4] = module self.update(settings) @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_HELPER, in_signature='s', out_signature='b') @dbus_handle_exceptions def queryModule(self, module, sender=None): # pylint: disable=W0613 module = dbus_to_python(module, str) log.debug1("%s.queryModule('%s')", self._log_prefix, module) settings = self.getSettings() return (settings[4] == module) # port @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_HELPER, out_signature='a(ss)') @dbus_handle_exceptions def getPorts(self, sender=None): # pylint: disable=W0613 log.debug1("%s.getPorts()", self._log_prefix) return self.getSettings()[5] @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_HELPER, in_signature='a(ss)') @dbus_handle_exceptions def setPorts(self, ports, sender=None): _ports = [ ] # convert embedded lists to tuples for port in dbus_to_python(ports, list): if isinstance(port, list): _ports.append(tuple(port)) else: _ports.append(port) ports = _ports log.debug1("%s.setPorts('[%s]')", self._log_prefix, ",".join("('%s, '%s')" % (port[0], port[1]) for port in ports)) self.parent.accessCheck(sender) settings = list(self.getSettings()) settings[5] = ports self.update(settings) @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_HELPER, in_signature='ss') @dbus_handle_exceptions def addPort(self, port, protocol, sender=None): port = dbus_to_python(port, str) protocol = dbus_to_python(protocol, str) log.debug1("%s.addPort('%s', '%s')", self._log_prefix, port, protocol) self.parent.accessCheck(sender) settings = list(self.getSettings()) if (port,protocol) in settings[5]: raise FirewallError(errors.ALREADY_ENABLED, "%s:%s" % (port, protocol)) settings[5].append((port,protocol)) self.update(settings) @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_HELPER, in_signature='ss') @dbus_handle_exceptions def removePort(self, port, protocol, sender=None): port = dbus_to_python(port, str) protocol = dbus_to_python(protocol, str) log.debug1("%s.removePort('%s', '%s')", self._log_prefix, port, protocol) self.parent.accessCheck(sender) settings = list(self.getSettings()) if (port,protocol) not in settings[5]: raise FirewallError(errors.NOT_ENABLED, "%s:%s" % (port, protocol)) settings[5].remove((port,protocol)) self.update(settings) @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_HELPER, in_signature='ss', out_signature='b') @dbus_handle_exceptions def queryPort(self, port, protocol, sender=None): # pylint: disable=W0613 port = dbus_to_python(port, str) protocol = dbus_to_python(protocol, str) log.debug1("%s.queryPort('%s', '%s')", self._log_prefix, port, protocol) return (port,protocol) in self.getSettings()[5]	17.04 KB	2026-03-10 23:36
📄 config_icmptype.py # -- coding: utf-8 -- # # Copyright (C) 2010-2016 Red Hat, Inc. # # Authors: # Thomas Woerner <twoerner@redhat.com> # # This program is free software; you can redistribute it and/or modify # it under the terms of the GNU General Public License as published by # the Free Software Foundation; either version 2 of the License, or # (at your option) any later version. # # This program is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU General Public License for more details. # # You should have received a copy of the GNU General Public License # along with this program. If not, see <http://www.gnu.org/licenses/>. import dbus import dbus.service from firewall import config from firewall.dbus_utils import dbus_to_python, \ dbus_introspection_prepare_properties, \ dbus_introspection_add_properties from firewall.core.io.icmptype import IcmpType from firewall.core.logger import log from firewall.server.dbus import DbusServiceObject from firewall.server.decorators import handle_exceptions, \ dbus_handle_exceptions, dbus_service_method, \ dbus_polkit_require_auth from firewall import errors from firewall.errors import FirewallError ############################################################################ # # class FirewallDConfigIcmpType # ############################################################################ class FirewallDConfigIcmpType(DbusServiceObject): """FirewallD main class""" persistent = True """ Make FirewallD persistent. """ default_polkit_auth_required = config.dbus.PK_ACTION_CONFIG """ Use PK_ACTION_INFO as a default """ @handle_exceptions def __init__(self, parent, conf, icmptype, item_id, args, kwargs): super(FirewallDConfigIcmpType, self).__init__(args, **kwargs) self.parent = parent self.config = conf self.obj = icmptype self.item_id = item_id self.busname = args[0] self.path = args[1] self._log_prefix = "config.icmptype.%d" % self.item_id dbus_introspection_prepare_properties( self, config.dbus.DBUS_INTERFACE_CONFIG_ICMPTYPE) @dbus_handle_exceptions def __del__(self): pass @dbus_handle_exceptions def unregister(self): self.remove_from_connection() # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # P R O P E R T I E S @dbus_handle_exceptions def _get_property(self, property_name): if property_name == "name": return dbus.String(self.obj.name) elif property_name == "filename": return dbus.String(self.obj.filename) elif property_name == "path": return dbus.String(self.obj.path) elif property_name == "default": return dbus.Boolean(self.obj.default) elif property_name == "builtin": return dbus.Boolean(self.obj.builtin) else: raise dbus.exceptions.DBusException( "org.freedesktop.DBus.Error.InvalidArgs: " "Property '%s' does not exist" % property_name) @dbus_service_method(dbus.PROPERTIES_IFACE, in_signature='ss', out_signature='v') @dbus_handle_exceptions def Get(self, interface_name, property_name, sender=None): # pylint: disable=W0613 # get a property interface_name = dbus_to_python(interface_name, str) property_name = dbus_to_python(property_name, str) log.debug1("%s.Get('%s', '%s')", self._log_prefix, interface_name, property_name) if interface_name != config.dbus.DBUS_INTERFACE_CONFIG_ICMPTYPE: raise dbus.exceptions.DBusException( "org.freedesktop.DBus.Error.UnknownInterface: " "Interface '%s' does not exist" % interface_name) return self._get_property(property_name) @dbus_service_method(dbus.PROPERTIES_IFACE, in_signature='s', out_signature='a{sv}') @dbus_handle_exceptions def GetAll(self, interface_name, sender=None): # pylint: disable=W0613 interface_name = dbus_to_python(interface_name, str) log.debug1("%s.GetAll('%s')", self._log_prefix, interface_name) if interface_name != config.dbus.DBUS_INTERFACE_CONFIG_ICMPTYPE: raise dbus.exceptions.DBusException( "org.freedesktop.DBus.Error.UnknownInterface: " "Interface '%s' does not exist" % interface_name) ret = { } for x in [ "name", "filename", "path", "default", "builtin" ]: ret[x] = self._get_property(x) return dbus.Dictionary(ret, signature="sv") @dbus_polkit_require_auth(config.dbus.PK_ACTION_CONFIG) @dbus_service_method(dbus.PROPERTIES_IFACE, in_signature='ssv') @dbus_handle_exceptions def Set(self, interface_name, property_name, new_value, sender=None): interface_name = dbus_to_python(interface_name, str) property_name = dbus_to_python(property_name, str) new_value = dbus_to_python(new_value) log.debug1("%s.Set('%s', '%s', '%s')", self._log_prefix, interface_name, property_name, new_value) self.parent.accessCheck(sender) if interface_name != config.dbus.DBUS_INTERFACE_CONFIG_ICMPTYPE: raise dbus.exceptions.DBusException( "org.freedesktop.DBus.Error.UnknownInterface: " "Interface '%s' does not exist" % interface_name) raise dbus.exceptions.DBusException( "org.freedesktop.DBus.Error.PropertyReadOnly: " "Property '%s' is read-only" % property_name) @dbus.service.signal(dbus.PROPERTIES_IFACE, signature='sa{sv}as') def PropertiesChanged(self, interface_name, changed_properties, invalidated_properties): interface_name = dbus_to_python(interface_name, str) changed_properties = dbus_to_python(changed_properties) invalidated_properties = dbus_to_python(invalidated_properties) log.debug1("%s.PropertiesChanged('%s', '%s', '%s')", self._log_prefix, interface_name, changed_properties, invalidated_properties) @dbus_polkit_require_auth(config.dbus.PK_ACTION_INFO) @dbus_service_method(dbus.INTROSPECTABLE_IFACE, out_signature='s') @dbus_handle_exceptions def Introspect(self, sender=None): # pylint: disable=W0613 log.debug2("%s.Introspect()", self._log_prefix) data = super(FirewallDConfigIcmpType, self).Introspect( self.path, self.busname.get_bus()) return dbus_introspection_add_properties( self, data, config.dbus.DBUS_INTERFACE_CONFIG_ICMPTYPE) # S E T T I N G S @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_ICMPTYPE, out_signature=IcmpType.DBUS_SIGNATURE) @dbus_handle_exceptions def getSettings(self, sender=None): # pylint: disable=W0613 """get settings for icmptype """ log.debug1("%s.getSettings()", self._log_prefix) return self.config.get_icmptype_config(self.obj) @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_ICMPTYPE, in_signature=IcmpType.DBUS_SIGNATURE) @dbus_handle_exceptions def update(self, settings, sender=None): """update settings for icmptype """ settings = dbus_to_python(settings) log.debug1("%s.update('...')", self._log_prefix) self.parent.accessCheck(sender) self.obj = self.config.set_icmptype_config(self.obj, settings) self.Updated(self.obj.name) @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_ICMPTYPE) @dbus_handle_exceptions def loadDefaults(self, sender=None): """load default settings for builtin icmptype """ log.debug1("%s.loadDefaults()", self._log_prefix) self.parent.accessCheck(sender) self.obj = self.config.load_icmptype_defaults(self.obj) self.Updated(self.obj.name) @dbus.service.signal(config.dbus.DBUS_INTERFACE_CONFIG_ICMPTYPE, signature='s') @dbus_handle_exceptions def Updated(self, name): log.debug1("%s.Updated('%s')" % (self._log_prefix, name)) # R E M O V E @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_ICMPTYPE) @dbus_handle_exceptions def remove(self, sender=None): """remove icmptype """ log.debug1("%s.removeIcmpType()", self._log_prefix) self.parent.accessCheck(sender) self.config.remove_icmptype(self.obj) self.parent.removeIcmpType(self.obj) @dbus.service.signal(config.dbus.DBUS_INTERFACE_CONFIG_ICMPTYPE, signature='s') @dbus_handle_exceptions def Removed(self, name): log.debug1("%s.Removed('%s')" % (self._log_prefix, name)) # R E N A M E @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_ICMPTYPE, in_signature='s') @dbus_handle_exceptions def rename(self, name, sender=None): """rename icmptype """ name = dbus_to_python(name, str) log.debug1("%s.rename('%s')", self._log_prefix, name) self.parent.accessCheck(sender) self.obj = self.config.rename_icmptype(self.obj, name) self.Renamed(name) @dbus.service.signal(config.dbus.DBUS_INTERFACE_CONFIG_ICMPTYPE, signature='s') @dbus_handle_exceptions def Renamed(self, name): log.debug1("%s.Renamed('%s')" % (self._log_prefix, name)) # version @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_ICMPTYPE, out_signature='s') @dbus_handle_exceptions def getVersion(self, sender=None): # pylint: disable=W0613 log.debug1("%s.getVersion()", self._log_prefix) return self.getSettings()[0] @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_ICMPTYPE, in_signature='s') @dbus_handle_exceptions def setVersion(self, version, sender=None): version = dbus_to_python(version, str) log.debug1("%s.setVersion('%s')", self._log_prefix, version) self.parent.accessCheck(sender) settings = list(self.getSettings()) settings[0] = version self.update(settings) # short @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_ICMPTYPE, out_signature='s') @dbus_handle_exceptions def getShort(self, sender=None): # pylint: disable=W0613 log.debug1("%s.getShort()", self._log_prefix) return self.getSettings()[1] @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_ICMPTYPE, in_signature='s') @dbus_handle_exceptions def setShort(self, short, sender=None): short = dbus_to_python(short, str) log.debug1("%s.setShort('%s')", self._log_prefix, short) self.parent.accessCheck(sender) settings = list(self.getSettings()) settings[1] = short self.update(settings) # description @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_ICMPTYPE, out_signature='s') @dbus_handle_exceptions def getDescription(self, sender=None): # pylint: disable=W0613 log.debug1("%s.getDescription()", self._log_prefix) return self.getSettings()[2] @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_ICMPTYPE, in_signature='s') @dbus_handle_exceptions def setDescription(self, description, sender=None): description = dbus_to_python(description, str) log.debug1("%s.setDescription('%s')", self._log_prefix, description) self.parent.accessCheck(sender) settings = list(self.getSettings()) settings[2] = description self.update(settings) # destination @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_ICMPTYPE, out_signature='as') @dbus_handle_exceptions def getDestinations(self, sender=None): # pylint: disable=W0613 log.debug1("%s.getDestinations()", self._log_prefix) return sorted(self.getSettings()[3]) @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_ICMPTYPE, in_signature='as') @dbus_handle_exceptions def setDestinations(self, destinations, sender=None): destinations = dbus_to_python(destinations, list) log.debug1("%s.setDestinations('[%s]')", self._log_prefix, ",".join(destinations)) self.parent.accessCheck(sender) settings = list(self.getSettings()) settings[3] = destinations self.update(settings) @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_ICMPTYPE, in_signature='s') @dbus_handle_exceptions def addDestination(self, destination, sender=None): destination = dbus_to_python(destination, str) log.debug1("%s.addDestination('%s')", self._log_prefix, destination) self.parent.accessCheck(sender) settings = list(self.getSettings()) if destination in settings[3]: raise FirewallError(errors.ALREADY_ENABLED, destination) settings[3].append(destination) self.update(settings) @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_ICMPTYPE, in_signature='s') @dbus_handle_exceptions def removeDestination(self, destination, sender=None): destination = dbus_to_python(destination, str) log.debug1("%s.removeDestination('%s')", self._log_prefix, destination) self.parent.accessCheck(sender) settings = list(self.getSettings()) if settings[3]: if destination not in settings[3]: raise FirewallError(errors.NOT_ENABLED, destination) else: settings[3].remove(destination) else: # empty means all settings[3] = list(set(['ipv4', 'ipv6']) - set([destination])) self.update(settings) @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_ICMPTYPE, in_signature='s', out_signature='b') @dbus_handle_exceptions def queryDestination(self, destination, sender=None): # pylint: disable=W0613 destination = dbus_to_python(destination, str) log.debug1("%s.queryDestination('%s')", self._log_prefix, destination) settings = self.getSettings() # empty means all return (not settings[3] or destination in settings[3])	14.49 KB	2026-03-10 23:36
📄 config_ipset.py # -- coding: utf-8 -- # # Copyright (C) 2015-2016 Red Hat, Inc. # # Authors: # Thomas Woerner <twoerner@redhat.com> # # This program is free software; you can redistribute it and/or modify # it under the terms of the GNU General Public License as published by # the Free Software Foundation; either version 2 of the License, or # (at your option) any later version. # # This program is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU General Public License for more details. # # You should have received a copy of the GNU General Public License # along with this program. If not, see <http://www.gnu.org/licenses/>. import dbus import dbus.service from firewall import config from firewall.dbus_utils import dbus_to_python, \ dbus_introspection_prepare_properties, \ dbus_introspection_add_properties from firewall.core.io.ipset import IPSet from firewall.core.ipset import IPSET_TYPES, normalize_ipset_entry, \ check_entry_overlaps_existing, \ check_for_overlapping_entries from firewall.core.logger import log from firewall.server.dbus import DbusServiceObject from firewall.server.decorators import handle_exceptions, \ dbus_handle_exceptions, dbus_service_method, \ dbus_polkit_require_auth from firewall import errors from firewall.errors import FirewallError ############################################################################ # # class FirewallDConfigIPSet # ############################################################################ class FirewallDConfigIPSet(DbusServiceObject): """FirewallD main class""" persistent = True """ Make FirewallD persistent. """ default_polkit_auth_required = config.dbus.PK_ACTION_CONFIG """ Use PK_ACTION_INFO as a default """ @handle_exceptions def __init__(self, parent, conf, ipset, item_id, args, kwargs): super(FirewallDConfigIPSet, self).__init__(args, **kwargs) self.parent = parent self.config = conf self.obj = ipset self.item_id = item_id self.busname = args[0] self.path = args[1] self._log_prefix = "config.ipset.%d" % self.item_id dbus_introspection_prepare_properties( self, config.dbus.DBUS_INTERFACE_CONFIG_IPSET) @dbus_handle_exceptions def __del__(self): pass @dbus_handle_exceptions def unregister(self): self.remove_from_connection() # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # P R O P E R T I E S @dbus_handle_exceptions def _get_property(self, property_name): if property_name == "name": return dbus.String(self.obj.name) elif property_name == "filename": return dbus.String(self.obj.filename) elif property_name == "path": return dbus.String(self.obj.path) elif property_name == "default": return dbus.Boolean(self.obj.default) elif property_name == "builtin": return dbus.Boolean(self.obj.builtin) else: raise dbus.exceptions.DBusException( "org.freedesktop.DBus.Error.InvalidArgs: " "Property '%s' does not exist" % property_name) @dbus_service_method(dbus.PROPERTIES_IFACE, in_signature='ss', out_signature='v') @dbus_handle_exceptions def Get(self, interface_name, property_name, sender=None): # pylint: disable=W0613 # get a property interface_name = dbus_to_python(interface_name, str) property_name = dbus_to_python(property_name, str) log.debug1("%s.Get('%s', '%s')", self._log_prefix, interface_name, property_name) if interface_name != config.dbus.DBUS_INTERFACE_CONFIG_IPSET: raise dbus.exceptions.DBusException( "org.freedesktop.DBus.Error.UnknownInterface: " "Interface '%s' does not exist" % interface_name) return self._get_property(property_name) @dbus_service_method(dbus.PROPERTIES_IFACE, in_signature='s', out_signature='a{sv}') @dbus_handle_exceptions def GetAll(self, interface_name, sender=None): # pylint: disable=W0613 interface_name = dbus_to_python(interface_name, str) log.debug1("%s.GetAll('%s')", self._log_prefix, interface_name) if interface_name != config.dbus.DBUS_INTERFACE_CONFIG_IPSET: raise dbus.exceptions.DBusException( "org.freedesktop.DBus.Error.UnknownInterface: " "Interface '%s' does not exist" % interface_name) ret = { } for x in [ "name", "filename", "path", "default", "builtin" ]: ret[x] = self._get_property(x) return dbus.Dictionary(ret, signature="sv") @dbus_polkit_require_auth(config.dbus.PK_ACTION_CONFIG) @dbus_service_method(dbus.PROPERTIES_IFACE, in_signature='ssv') @dbus_handle_exceptions def Set(self, interface_name, property_name, new_value, sender=None): interface_name = dbus_to_python(interface_name, str) property_name = dbus_to_python(property_name, str) new_value = dbus_to_python(new_value) log.debug1("%s.Set('%s', '%s', '%s')", self._log_prefix, interface_name, property_name, new_value) self.parent.accessCheck(sender) if interface_name != config.dbus.DBUS_INTERFACE_CONFIG_IPSET: raise dbus.exceptions.DBusException( "org.freedesktop.DBus.Error.UnknownInterface: " "Interface '%s' does not exist" % interface_name) raise dbus.exceptions.DBusException( "org.freedesktop.DBus.Error.PropertyReadOnly: " "Property '%s' is read-only" % property_name) @dbus.service.signal(dbus.PROPERTIES_IFACE, signature='sa{sv}as') def PropertiesChanged(self, interface_name, changed_properties, invalidated_properties): interface_name = dbus_to_python(interface_name, str) changed_properties = dbus_to_python(changed_properties) invalidated_properties = dbus_to_python(invalidated_properties) log.debug1("%s.PropertiesChanged('%s', '%s', '%s')", self._log_prefix, interface_name, changed_properties, invalidated_properties) @dbus_polkit_require_auth(config.dbus.PK_ACTION_INFO) @dbus_service_method(dbus.INTROSPECTABLE_IFACE, out_signature='s') @dbus_handle_exceptions def Introspect(self, sender=None): # pylint: disable=W0613 log.debug2("%s.Introspect()", self._log_prefix) data = super(FirewallDConfigIPSet, self).Introspect( self.path, self.busname.get_bus()) return dbus_introspection_add_properties( self, data, config.dbus.DBUS_INTERFACE_CONFIG_IPSET) # S E T T I N G S @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_IPSET, out_signature=IPSet.DBUS_SIGNATURE) @dbus_handle_exceptions def getSettings(self, sender=None): # pylint: disable=W0613 """get settings for ipset """ log.debug1("%s.getSettings()", self._log_prefix) return self.config.get_ipset_config(self.obj) @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_IPSET, in_signature=IPSet.DBUS_SIGNATURE) @dbus_handle_exceptions def update(self, settings, sender=None): """update settings for ipset """ settings = dbus_to_python(settings) log.debug1("%s.update('...')", self._log_prefix) self.parent.accessCheck(sender) self.obj = self.config.set_ipset_config(self.obj, settings) self.Updated(self.obj.name) @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_IPSET) @dbus_handle_exceptions def loadDefaults(self, sender=None): """load default settings for builtin ipset """ log.debug1("%s.loadDefaults()", self._log_prefix) self.parent.accessCheck(sender) self.obj = self.config.load_ipset_defaults(self.obj) self.Updated(self.obj.name) #self.PropertiesChanged(config.dbus.DBUS_INTERFACE_CONFIG_IPSET, # { "default": True }, [ ]) @dbus.service.signal(config.dbus.DBUS_INTERFACE_CONFIG_IPSET, signature='s') @dbus_handle_exceptions def Updated(self, name): log.debug1("%s.Updated('%s')" % (self._log_prefix, name)) # R E M O V E @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_IPSET) @dbus_handle_exceptions def remove(self, sender=None): """remove ipset """ log.debug1("%s.remove()", self._log_prefix) self.parent.accessCheck(sender) self.config.remove_ipset(self.obj) self.parent.removeIPSet(self.obj) @dbus.service.signal(config.dbus.DBUS_INTERFACE_CONFIG_IPSET, signature='s') @dbus_handle_exceptions def Removed(self, name): log.debug1("%s.Removed('%s')" % (self._log_prefix, name)) # R E N A M E @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_IPSET, in_signature='s') @dbus_handle_exceptions def rename(self, name, sender=None): """rename ipset """ name = dbus_to_python(name, str) log.debug1("%s.rename('%s')", self._log_prefix, name) self.parent.accessCheck(sender) self.obj = self.config.rename_ipset(self.obj, name) self.Renamed(name) #self.PropertiesChanged(config.dbus.DBUS_INTERFACE_CONFIG_IPSET, # { "name": name }, [ ]) @dbus.service.signal(config.dbus.DBUS_INTERFACE_CONFIG_IPSET, signature='s') @dbus_handle_exceptions def Renamed(self, name): log.debug1("%s.Renamed('%s')" % (self._log_prefix, name)) # version @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_IPSET, out_signature='s') @dbus_handle_exceptions def getVersion(self, sender=None): # pylint: disable=W0613 log.debug1("%s.getVersion()", self._log_prefix) return self.getSettings()[0] @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_IPSET, in_signature='s') @dbus_handle_exceptions def setVersion(self, version, sender=None): version = dbus_to_python(version, str) log.debug1("%s.setVersion('%s')", self._log_prefix, version) self.parent.accessCheck(sender) settings = list(self.getSettings()) settings[0] = version self.update(settings) # short @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_IPSET, out_signature='s') @dbus_handle_exceptions def getShort(self, sender=None): # pylint: disable=W0613 log.debug1("%s.getShort()", self._log_prefix) return self.getSettings()[1] @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_IPSET, in_signature='s') @dbus_handle_exceptions def setShort(self, short, sender=None): short = dbus_to_python(short, str) log.debug1("%s.setShort('%s')", self._log_prefix, short) self.parent.accessCheck(sender) settings = list(self.getSettings()) settings[1] = short self.update(settings) # description @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_IPSET, out_signature='s') @dbus_handle_exceptions def getDescription(self, sender=None): # pylint: disable=W0613 log.debug1("%s.getDescription()", self._log_prefix) return self.getSettings()[2] @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_IPSET, in_signature='s') @dbus_handle_exceptions def setDescription(self, description, sender=None): description = dbus_to_python(description, str) log.debug1("%s.setDescription('%s')", self._log_prefix, description) self.parent.accessCheck(sender) settings = list(self.getSettings()) settings[2] = description self.update(settings) # type @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_IPSET, out_signature='s') @dbus_handle_exceptions def getType(self, sender=None): # pylint: disable=W0613 log.debug1("%s.getType()", self._log_prefix) return self.getSettings()[3] @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_IPSET, in_signature='s') @dbus_handle_exceptions def setType(self, ipset_type, sender=None): ipset_type = dbus_to_python(ipset_type, str) log.debug1("%s.setType('%s')", self._log_prefix, ipset_type) self.parent.accessCheck(sender) if ipset_type not in IPSET_TYPES: raise FirewallError(errors.INVALID_TYPE, ipset_type) settings = list(self.getSettings()) settings[3] = ipset_type self.update(settings) # options @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_IPSET, out_signature='a{ss}') @dbus_handle_exceptions def getOptions(self, sender=None): # pylint: disable=W0613 log.debug1("%s.getOptions()", self._log_prefix) return self.getSettings()[4] @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_IPSET, in_signature='a{ss}') @dbus_handle_exceptions def setOptions(self, options, sender=None): options = dbus_to_python(options, dict) log.debug1("%s.setOptions('[%s]')", self._log_prefix, repr(options)) self.parent.accessCheck(sender) settings = list(self.getSettings()) settings[4] = options self.update(settings) @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_IPSET, in_signature='ss') @dbus_handle_exceptions def addOption(self, key, value, sender=None): key = dbus_to_python(key, str) value = dbus_to_python(value, str) log.debug1("%s.addOption('%s', '%s')", self._log_prefix, key, value) self.parent.accessCheck(sender) settings = list(self.getSettings()) if key in settings[4] and settings[4][key] == value: raise FirewallError(errors.ALREADY_ENABLED, "'%s': '%s'" % (key, value)) settings[4][key] = value self.update(settings) @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_IPSET, in_signature='s') @dbus_handle_exceptions def removeOption(self, key, sender=None): key = dbus_to_python(key, str) log.debug1("%s.removeOption('%s')", self._log_prefix, key) self.parent.accessCheck(sender) settings = list(self.getSettings()) if key not in settings[4]: raise FirewallError(errors.NOT_ENABLED, key) del settings[4][key] self.update(settings) @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_IPSET, in_signature='ss', out_signature='b') @dbus_handle_exceptions def queryOption(self, key, value, sender=None): # pylint: disable=W0613 key = dbus_to_python(key, str) value = dbus_to_python(value, str) log.debug1("%s.queryOption('%s', '%s')", self._log_prefix, key, value) settings = list(self.getSettings()) return (key in settings[4] and settings[4][key] == value) # entries @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_IPSET, out_signature='as') @dbus_handle_exceptions def getEntries(self, sender=None): # pylint: disable=W0613 log.debug1("%s.getEntries()", self._log_prefix) return self.getSettings()[5] @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_IPSET, in_signature='as') @dbus_handle_exceptions def setEntries(self, entries, sender=None): entries = dbus_to_python(entries, list) check_for_overlapping_entries(entries) log.debug1("%s.setEntries('[%s]')", self._log_prefix, ",".join(entries)) self.parent.accessCheck(sender) settings = list(self.getSettings()) if "timeout" in settings[4] and settings[4]["timeout"] != "0": raise FirewallError(errors.IPSET_WITH_TIMEOUT) settings[5] = entries self.update(settings) @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_IPSET, in_signature='s') @dbus_handle_exceptions def addEntry(self, entry, sender=None): entry = dbus_to_python(entry, str) entry = normalize_ipset_entry(entry) log.debug1("%s.addEntry('%s')", self._log_prefix, entry) self.parent.accessCheck(sender) settings = list(self.getSettings()) if "timeout" in settings[4] and settings[4]["timeout"] != "0": raise FirewallError(errors.IPSET_WITH_TIMEOUT) if entry in settings[5]: raise FirewallError(errors.ALREADY_ENABLED, entry) check_entry_overlaps_existing(entry, settings[5]) settings[5].append(entry) self.update(settings) @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_IPSET, in_signature='s') @dbus_handle_exceptions def removeEntry(self, entry, sender=None): entry = dbus_to_python(entry, str) entry = normalize_ipset_entry(entry) log.debug1("%s.removeEntry('%s')", self._log_prefix, entry) self.parent.accessCheck(sender) settings = list(self.getSettings()) if "timeout" in settings[4] and settings[4]["timeout"] != "0": raise FirewallError(errors.IPSET_WITH_TIMEOUT) if entry not in settings[5]: raise FirewallError(errors.NOT_ENABLED, entry) settings[5].remove(entry) self.update(settings) @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_IPSET, in_signature='s', out_signature='b') @dbus_handle_exceptions def queryEntry(self, entry, sender=None): # pylint: disable=W0613 entry = dbus_to_python(entry, str) entry = normalize_ipset_entry(entry) log.debug1("%s.queryEntry('%s')", self._log_prefix, entry) settings = list(self.getSettings()) if "timeout" in settings[4] and settings[4]["timeout"] != "0": raise FirewallError(errors.IPSET_WITH_TIMEOUT) return entry in settings[5]	18.28 KB	2026-03-10 23:36
📄 config_policy.py # -- coding: utf-8 -- # # SPDX-License-Identifier: GPL-2.0-or-later import dbus import dbus.service from firewall import config from firewall.dbus_utils import dbus_to_python, \ dbus_introspection_prepare_properties, \ dbus_introspection_add_properties from firewall.core.logger import log from firewall.server.dbus import DbusServiceObject from firewall.server.decorators import handle_exceptions, \ dbus_handle_exceptions, dbus_service_method, \ dbus_polkit_require_auth class FirewallDConfigPolicy(DbusServiceObject): persistent = True default_polkit_auth_required = config.dbus.PK_ACTION_CONFIG @handle_exceptions def __init__(self, parent, conf, policy, item_id, args, kwargs): super(FirewallDConfigPolicy, self).__init__(args, **kwargs) self.parent = parent self.config = conf self.obj = policy self.item_id = item_id self.busname = args[0] self.path = args[1] self._log_prefix = "config.policy.%d" % self.item_id dbus_introspection_prepare_properties( self, config.dbus.DBUS_INTERFACE_CONFIG_POLICY) @dbus_handle_exceptions def __del__(self): pass @dbus_handle_exceptions def unregister(self): self.remove_from_connection() # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # P R O P E R T I E S @dbus_handle_exceptions def _get_property(self, property_name): if property_name == "name": return dbus.String(self.obj.name) elif property_name == "filename": return dbus.String(self.obj.filename) elif property_name == "path": return dbus.String(self.obj.path) elif property_name == "default": return dbus.Boolean(self.obj.default) elif property_name == "builtin": return dbus.Boolean(self.obj.builtin) else: raise dbus.exceptions.DBusException( "org.freedesktop.DBus.Error.InvalidArgs: " "Property '%s' does not exist" % property_name) @dbus_service_method(dbus.PROPERTIES_IFACE, in_signature='ss', out_signature='v') @dbus_handle_exceptions def Get(self, interface_name, property_name, sender=None): # get a property interface_name = dbus_to_python(interface_name, str) property_name = dbus_to_python(property_name, str) log.debug1("%s.Get('%s', '%s')", self._log_prefix, interface_name, property_name) if interface_name != config.dbus.DBUS_INTERFACE_CONFIG_POLICY: raise dbus.exceptions.DBusException( "org.freedesktop.DBus.Error.UnknownInterface: " "Interface '%s' does not exist" % interface_name) return self._get_property(property_name) @dbus_service_method(dbus.PROPERTIES_IFACE, in_signature='s', out_signature='a{sv}') @dbus_handle_exceptions def GetAll(self, interface_name, sender=None): interface_name = dbus_to_python(interface_name, str) log.debug1("%s.GetAll('%s')", self._log_prefix, interface_name) if interface_name != config.dbus.DBUS_INTERFACE_CONFIG_POLICY: raise dbus.exceptions.DBusException( "org.freedesktop.DBus.Error.UnknownInterface: " "Interface '%s' does not exist" % interface_name) ret = { } for x in [ "name", "filename", "path", "default", "builtin" ]: ret[x] = self._get_property(x) return dbus.Dictionary(ret, signature="sv") @dbus_polkit_require_auth(config.dbus.PK_ACTION_CONFIG) @dbus_service_method(dbus.PROPERTIES_IFACE, in_signature='ssv') @dbus_handle_exceptions def Set(self, interface_name, property_name, new_value, sender=None): interface_name = dbus_to_python(interface_name, str) property_name = dbus_to_python(property_name, str) new_value = dbus_to_python(new_value) log.debug1("%s.Set('%s', '%s', '%s')", self._log_prefix, interface_name, property_name, new_value) self.parent.accessCheck(sender) if interface_name != config.dbus.DBUS_INTERFACE_CONFIG_POLICY: raise dbus.exceptions.DBusException( "org.freedesktop.DBus.Error.UnknownInterface: " "Interface '%s' does not exist" % interface_name) raise dbus.exceptions.DBusException( "org.freedesktop.DBus.Error.PropertyReadOnly: " "Property '%s' is read-only" % property_name) @dbus.service.signal(dbus.PROPERTIES_IFACE, signature='sa{sv}as') def PropertiesChanged(self, interface_name, changed_properties, invalidated_properties): interface_name = dbus_to_python(interface_name, str) changed_properties = dbus_to_python(changed_properties) invalidated_properties = dbus_to_python(invalidated_properties) log.debug1("%s.PropertiesChanged('%s', '%s', '%s')", self._log_prefix, interface_name, changed_properties, invalidated_properties) @dbus_polkit_require_auth(config.dbus.PK_ACTION_INFO) @dbus_service_method(dbus.INTROSPECTABLE_IFACE, out_signature='s') @dbus_handle_exceptions def Introspect(self, sender=None): log.debug2("%s.Introspect()", self._log_prefix) data = super(FirewallDConfigPolicy, self).Introspect( self.path, self.busname.get_bus()) return dbus_introspection_add_properties( self, data, config.dbus.DBUS_INTERFACE_CONFIG_POLICY) # S E T T I N G S @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_POLICY, out_signature="a{sv}") @dbus_handle_exceptions def getSettings(self, sender=None): """get settings for policy """ log.debug1("%s.getSettings()", self._log_prefix) settings = self.config.get_policy_object_config_dict(self.obj) return settings @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_POLICY, in_signature="a{sv}") @dbus_handle_exceptions def update(self, settings, sender=None): """update settings for policy """ settings = dbus_to_python(settings) log.debug1("%s.update('...')", self._log_prefix) self.parent.accessCheck(sender) self.obj = self.config.set_policy_object_config_dict(self.obj, settings) self.Updated(self.obj.name) @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_POLICY) @dbus_handle_exceptions def loadDefaults(self, sender=None): """load default settings for builtin policy """ log.debug1("%s.loadDefaults()", self._log_prefix) self.parent.accessCheck(sender) self.obj = self.config.load_policy_object_defaults(self.obj) self.Updated(self.obj.name) @dbus.service.signal(config.dbus.DBUS_INTERFACE_CONFIG_POLICY, signature='s') @dbus_handle_exceptions def Updated(self, name): log.debug1("%s.Updated('%s')" % (self._log_prefix, name)) # R E M O V E @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_POLICY) @dbus_handle_exceptions def remove(self, sender=None): """remove policy """ log.debug1("%s.removePolicy()", self._log_prefix) self.parent.accessCheck(sender) self.config.remove_policy_object(self.obj) self.parent.removePolicy(self.obj) @dbus.service.signal(config.dbus.DBUS_INTERFACE_CONFIG_POLICY, signature='s') @dbus_handle_exceptions def Removed(self, name): log.debug1("%s.Removed('%s')" % (self._log_prefix, name)) # R E N A M E @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_POLICY, in_signature='s') @dbus_handle_exceptions def rename(self, name, sender=None): """rename policy """ name = dbus_to_python(name, str) log.debug1("%s.rename('%s')", self._log_prefix, name) self.parent.accessCheck(sender) self.obj = self.config.rename_policy_object(self.obj, name) self.Renamed(name) @dbus.service.signal(config.dbus.DBUS_INTERFACE_CONFIG_POLICY, signature='s') @dbus_handle_exceptions def Renamed(self, name): log.debug1("%s.Renamed('%s')" % (self._log_prefix, name))	8.2 KB	2026-03-10 23:36
📄 config_service.py # -- coding: utf-8 -- # # Copyright (C) 2010-2016 Red Hat, Inc. # # Authors: # Thomas Woerner <twoerner@redhat.com> # # This program is free software; you can redistribute it and/or modify # it under the terms of the GNU General Public License as published by # the Free Software Foundation; either version 2 of the License, or # (at your option) any later version. # # This program is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU General Public License for more details. # # You should have received a copy of the GNU General Public License # along with this program. If not, see <http://www.gnu.org/licenses/>. import dbus import dbus.service from firewall import config from firewall.dbus_utils import dbus_to_python, \ dbus_introspection_prepare_properties, \ dbus_introspection_add_properties from firewall.core.logger import log from firewall.server.dbus import DbusServiceObject from firewall.server.decorators import handle_exceptions, \ dbus_handle_exceptions, dbus_service_method, \ dbus_polkit_require_auth from firewall import errors from firewall.errors import FirewallError ############################################################################ # # class FirewallDConfig # ############################################################################ class FirewallDConfigService(DbusServiceObject): """FirewallD main class""" persistent = True """ Make FirewallD persistent. """ default_polkit_auth_required = config.dbus.PK_ACTION_CONFIG """ Use PK_ACTION_INFO as a default """ @handle_exceptions def __init__(self, parent, conf, service, item_id, args, kwargs): super(FirewallDConfigService, self).__init__(args, **kwargs) self.parent = parent self.config = conf self.obj = service self.item_id = item_id self.busname = args[0] self.path = args[1] self._log_prefix = "config.service.%d" % self.item_id dbus_introspection_prepare_properties( self, config.dbus.DBUS_INTERFACE_CONFIG_SERVICE) @dbus_handle_exceptions def __del__(self): pass @dbus_handle_exceptions def unregister(self): self.remove_from_connection() # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # P R O P E R T I E S @dbus_handle_exceptions def _get_property(self, property_name): if property_name == "name": return dbus.String(self.obj.name) elif property_name == "filename": return dbus.String(self.obj.filename) elif property_name == "path": return dbus.String(self.obj.path) elif property_name == "default": return dbus.Boolean(self.obj.default) elif property_name == "builtin": return dbus.Boolean(self.obj.builtin) else: raise dbus.exceptions.DBusException( "org.freedesktop.DBus.Error.InvalidArgs: " "Property '%s' does not exist" % property_name) @dbus_service_method(dbus.PROPERTIES_IFACE, in_signature='ss', out_signature='v') @dbus_handle_exceptions def Get(self, interface_name, property_name, sender=None): # pylint: disable=W0613 # get a property interface_name = dbus_to_python(interface_name, str) property_name = dbus_to_python(property_name, str) log.debug1("%s.Get('%s', '%s')", self._log_prefix, interface_name, property_name) if interface_name != config.dbus.DBUS_INTERFACE_CONFIG_SERVICE: raise dbus.exceptions.DBusException( "org.freedesktop.DBus.Error.UnknownInterface: " "Interface '%s' does not exist" % interface_name) return self._get_property(property_name) @dbus_service_method(dbus.PROPERTIES_IFACE, in_signature='s', out_signature='a{sv}') @dbus_handle_exceptions def GetAll(self, interface_name, sender=None): # pylint: disable=W0613 interface_name = dbus_to_python(interface_name, str) log.debug1("%s.GetAll('%s')", self._log_prefix, interface_name) if interface_name != config.dbus.DBUS_INTERFACE_CONFIG_SERVICE: raise dbus.exceptions.DBusException( "org.freedesktop.DBus.Error.UnknownInterface: " "Interface '%s' does not exist" % interface_name) ret = { } for x in [ "name", "filename", "path", "default", "builtin" ]: ret[x] = self._get_property(x) return dbus.Dictionary(ret, signature="sv") @dbus_polkit_require_auth(config.dbus.PK_ACTION_CONFIG) @dbus_service_method(dbus.PROPERTIES_IFACE, in_signature='ssv') @dbus_handle_exceptions def Set(self, interface_name, property_name, new_value, sender=None): interface_name = dbus_to_python(interface_name, str) property_name = dbus_to_python(property_name, str) new_value = dbus_to_python(new_value) log.debug1("%s.Set('%s', '%s', '%s')", self._log_prefix, interface_name, property_name, new_value) self.parent.accessCheck(sender) if interface_name != config.dbus.DBUS_INTERFACE_CONFIG_SERVICE: raise dbus.exceptions.DBusException( "org.freedesktop.DBus.Error.UnknownInterface: " "Interface '%s' does not exist" % interface_name) raise dbus.exceptions.DBusException( "org.freedesktop.DBus.Error.PropertyReadOnly: " "Property '%s' is read-only" % property_name) @dbus.service.signal(dbus.PROPERTIES_IFACE, signature='sa{sv}as') def PropertiesChanged(self, interface_name, changed_properties, invalidated_properties): interface_name = dbus_to_python(interface_name, str) changed_properties = dbus_to_python(changed_properties) invalidated_properties = dbus_to_python(invalidated_properties) log.debug1("%s.PropertiesChanged('%s', '%s', '%s')", self._log_prefix, interface_name, changed_properties, invalidated_properties) @dbus_polkit_require_auth(config.dbus.PK_ACTION_INFO) @dbus_service_method(dbus.INTROSPECTABLE_IFACE, out_signature='s') @dbus_handle_exceptions def Introspect(self, sender=None): # pylint: disable=W0613 log.debug2("%s.Introspect()", self._log_prefix) data = super(FirewallDConfigService, self).Introspect( self.path, self.busname.get_bus()) return dbus_introspection_add_properties( self, data, config.dbus.DBUS_INTERFACE_CONFIG_SERVICE) # S E T T I N G S @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_SERVICE, out_signature='(sssa(ss)asa{ss}asa(ss))') @dbus_handle_exceptions def getSettings(self, sender=None): # pylint: disable=W0613 """get settings for service """ log.debug1("%s.getSettings()", self._log_prefix) return self.config.get_service_config(self.obj) @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_SERVICE, out_signature='a{sv}') @dbus_handle_exceptions def getSettings2(self, sender=None): """get settings for service """ log.debug1("%s.getSettings2()", self._log_prefix) return self.config.get_service_config_dict(self.obj) @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_SERVICE, in_signature='(sssa(ss)asa{ss}asa(ss))') @dbus_handle_exceptions def update(self, settings, sender=None): """update settings for service """ settings = dbus_to_python(settings) log.debug1("%s.update('...')", self._log_prefix) self.parent.accessCheck(sender) self.obj = self.config.set_service_config(self.obj, settings) self.Updated(self.obj.name) @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_SERVICE, in_signature='a{sv}') @dbus_handle_exceptions def update2(self, settings, sender=None): settings = dbus_to_python(settings) log.debug1("%s.update2('...')", self._log_prefix) self.parent.accessCheck(sender) self.obj = self.config.set_service_config_dict(self.obj, settings) self.Updated(self.obj.name) @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_SERVICE) @dbus_handle_exceptions def loadDefaults(self, sender=None): """load default settings for builtin service """ log.debug1("%s.loadDefaults()", self._log_prefix) self.parent.accessCheck(sender) self.obj = self.config.load_service_defaults(self.obj) self.Updated(self.obj.name) @dbus.service.signal(config.dbus.DBUS_INTERFACE_CONFIG_SERVICE, signature='s') @dbus_handle_exceptions def Updated(self, name): log.debug1("%s.Updated('%s')" % (self._log_prefix, name)) # R E M O V E @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_SERVICE) @dbus_handle_exceptions def remove(self, sender=None): """remove service """ log.debug1("%s.removeService()", self._log_prefix) self.parent.accessCheck(sender) self.config.remove_service(self.obj) self.parent.removeService(self.obj) @dbus.service.signal(config.dbus.DBUS_INTERFACE_CONFIG_SERVICE, signature='s') @dbus_handle_exceptions def Removed(self, name): log.debug1("%s.Removed('%s')" % (self._log_prefix, name)) # R E N A M E @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_SERVICE, in_signature='s') @dbus_handle_exceptions def rename(self, name, sender=None): """rename service """ name = dbus_to_python(name, str) log.debug1("%s.rename('%s')", self._log_prefix, name) self.parent.accessCheck(sender) self.obj = self.config.rename_service(self.obj, name) self.Renamed(name) @dbus.service.signal(config.dbus.DBUS_INTERFACE_CONFIG_SERVICE, signature='s') @dbus_handle_exceptions def Renamed(self, name): log.debug1("%s.Renamed('%s')" % (self._log_prefix, name)) # version @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_SERVICE, out_signature='s') @dbus_handle_exceptions def getVersion(self, sender=None): # pylint: disable=W0613 log.debug1("%s.getVersion()", self._log_prefix) return self.getSettings()[0] @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_SERVICE, in_signature='s') @dbus_handle_exceptions def setVersion(self, version, sender=None): version = dbus_to_python(version, str) log.debug1("%s.setVersion('%s')", self._log_prefix, version) self.parent.accessCheck(sender) settings = list(self.getSettings()) settings[0] = version self.update(settings) # short @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_SERVICE, out_signature='s') @dbus_handle_exceptions def getShort(self, sender=None): # pylint: disable=W0613 log.debug1("%s.getShort()", self._log_prefix) return self.getSettings()[1] @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_SERVICE, in_signature='s') @dbus_handle_exceptions def setShort(self, short, sender=None): short = dbus_to_python(short, str) log.debug1("%s.setShort('%s')", self._log_prefix, short) self.parent.accessCheck(sender) settings = list(self.getSettings()) settings[1] = short self.update(settings) # description @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_SERVICE, out_signature='s') @dbus_handle_exceptions def getDescription(self, sender=None): # pylint: disable=W0613 log.debug1("%s.getDescription()", self._log_prefix) return self.getSettings()[2] @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_SERVICE, in_signature='s') @dbus_handle_exceptions def setDescription(self, description, sender=None): description = dbus_to_python(description, str) log.debug1("%s.setDescription('%s')", self._log_prefix, description) self.parent.accessCheck(sender) settings = list(self.getSettings()) settings[2] = description self.update(settings) # port @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_SERVICE, out_signature='a(ss)') @dbus_handle_exceptions def getPorts(self, sender=None): # pylint: disable=W0613 log.debug1("%s.getPorts()", self._log_prefix) return self.getSettings()[3] @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_SERVICE, in_signature='a(ss)') @dbus_handle_exceptions def setPorts(self, ports, sender=None): _ports = [ ] # convert embedded lists to tuples for port in dbus_to_python(ports, list): if isinstance(port, list): _ports.append(tuple(port)) else: _ports.append(port) ports = _ports log.debug1("%s.setPorts('[%s]')", self._log_prefix, ",".join("('%s, '%s')" % (port[0], port[1]) for port in ports)) self.parent.accessCheck(sender) settings = list(self.getSettings()) settings[3] = ports self.update(settings) @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_SERVICE, in_signature='ss') @dbus_handle_exceptions def addPort(self, port, protocol, sender=None): port = dbus_to_python(port, str) protocol = dbus_to_python(protocol, str) log.debug1("%s.addPort('%s', '%s')", self._log_prefix, port, protocol) self.parent.accessCheck(sender) settings = list(self.getSettings()) if (port,protocol) in settings[3]: raise FirewallError(errors.ALREADY_ENABLED, "%s:%s" % (port, protocol)) settings[3].append((port,protocol)) self.update(settings) @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_SERVICE, in_signature='ss') @dbus_handle_exceptions def removePort(self, port, protocol, sender=None): port = dbus_to_python(port, str) protocol = dbus_to_python(protocol, str) log.debug1("%s.removePort('%s', '%s')", self._log_prefix, port, protocol) self.parent.accessCheck(sender) settings = list(self.getSettings()) if (port,protocol) not in settings[3]: raise FirewallError(errors.NOT_ENABLED, "%s:%s" % (port, protocol)) settings[3].remove((port,protocol)) self.update(settings) @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_SERVICE, in_signature='ss', out_signature='b') @dbus_handle_exceptions def queryPort(self, port, protocol, sender=None): # pylint: disable=W0613 port = dbus_to_python(port, str) protocol = dbus_to_python(protocol, str) log.debug1("%s.queryPort('%s', '%s')", self._log_prefix, port, protocol) return (port,protocol) in self.getSettings()[3] # protocol @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_SERVICE, out_signature='as') @dbus_handle_exceptions def getProtocols(self, sender=None): # pylint: disable=W0613 log.debug1("%s.getProtocols()", self._log_prefix) return self.getSettings()[6] @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_SERVICE, in_signature='as') @dbus_handle_exceptions def setProtocols(self, protocols, sender=None): protocols = dbus_to_python(protocols, list) log.debug1("%s.setProtocols('[%s]')", self._log_prefix, ",".join(protocols)) self.parent.accessCheck(sender) settings = list(self.getSettings()) settings[6] = protocols self.update(settings) @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_SERVICE, in_signature='s') @dbus_handle_exceptions def addProtocol(self, protocol, sender=None): protocol = dbus_to_python(protocol, str) log.debug1("%s.addProtocol('%s')", self._log_prefix, protocol) self.parent.accessCheck(sender) settings = list(self.getSettings()) if protocol in settings[6]: raise FirewallError(errors.ALREADY_ENABLED, protocol) settings[6].append(protocol) self.update(settings) @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_SERVICE, in_signature='s') @dbus_handle_exceptions def removeProtocol(self, protocol, sender=None): protocol = dbus_to_python(protocol, str) log.debug1("%s.removeProtocol('%s')", self._log_prefix, protocol) self.parent.accessCheck(sender) settings = list(self.getSettings()) if protocol not in settings[6]: raise FirewallError(errors.NOT_ENABLED, protocol) settings[6].remove(protocol) self.update(settings) @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_SERVICE, in_signature='s', out_signature='b') @dbus_handle_exceptions def queryProtocol(self, protocol, sender=None): # pylint: disable=W0613 protocol = dbus_to_python(protocol, str) log.debug1("%s.queryProtocol(%s')", self._log_prefix, protocol) return protocol in self.getSettings()[6] # source port @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_SERVICE, out_signature='a(ss)') @dbus_handle_exceptions def getSourcePorts(self, sender=None): # pylint: disable=W0613 log.debug1("%s.getSourcePorts()", self._log_prefix) return self.getSettings()[7] @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_SERVICE, in_signature='a(ss)') @dbus_handle_exceptions def setSourcePorts(self, ports, sender=None): _ports = [ ] # convert embedded lists to tuples for port in dbus_to_python(ports, list): if isinstance(port, list): _ports.append(tuple(port)) else: _ports.append(port) ports = _ports log.debug1("%s.setSourcePorts('[%s]')", self._log_prefix, ",".join("('%s, '%s')" % (port[0], port[1]) for port in ports)) self.parent.accessCheck(sender) settings = list(self.getSettings()) settings[7] = ports self.update(settings) @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_SERVICE, in_signature='ss') @dbus_handle_exceptions def addSourcePort(self, port, protocol, sender=None): port = dbus_to_python(port, str) protocol = dbus_to_python(protocol, str) log.debug1("%s.addSourcePort('%s', '%s')", self._log_prefix, port, protocol) self.parent.accessCheck(sender) settings = list(self.getSettings()) if (port,protocol) in settings[7]: raise FirewallError(errors.ALREADY_ENABLED, "%s:%s" % (port, protocol)) settings[7].append((port,protocol)) self.update(settings) @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_SERVICE, in_signature='ss') @dbus_handle_exceptions def removeSourcePort(self, port, protocol, sender=None): port = dbus_to_python(port, str) protocol = dbus_to_python(protocol, str) log.debug1("%s.removeSourcePort('%s', '%s')", self._log_prefix, port, protocol) self.parent.accessCheck(sender) settings = list(self.getSettings()) if (port,protocol) not in settings[7]: raise FirewallError(errors.NOT_ENABLED, "%s:%s" % (port, protocol)) settings[7].remove((port,protocol)) self.update(settings) @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_SERVICE, in_signature='ss', out_signature='b') @dbus_handle_exceptions def querySourcePort(self, port, protocol, sender=None): # pylint: disable=W0613 port = dbus_to_python(port, str) protocol = dbus_to_python(protocol, str) log.debug1("%s.querySourcePort('%s', '%s')", self._log_prefix, port, protocol) return (port,protocol) in self.getSettings()[7] # module @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_SERVICE, out_signature='as') @dbus_handle_exceptions def getModules(self, sender=None): # pylint: disable=W0613 log.debug1("%s.getModules()", self._log_prefix) return self.getSettings()[4] @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_SERVICE, in_signature='as') @dbus_handle_exceptions def setModules(self, modules, sender=None): modules = dbus_to_python(modules, list) _modules = [ ] for module in modules: if module.startswith("nf_conntrack_"): module = module.replace("nf_conntrack_", "") if "_" in module: module = module.replace("_", "-") _modules.append(module) modules = _modules log.debug1("%s.setModules('[%s]')", self._log_prefix, ",".join(modules)) self.parent.accessCheck(sender) settings = list(self.getSettings()) settings[4] = modules self.update(settings) @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_SERVICE, in_signature='s') @dbus_handle_exceptions def addModule(self, module, sender=None): module = dbus_to_python(module, str) if module.startswith("nf_conntrack_"): module = module.replace("nf_conntrack_", "") if "_" in module: module = module.replace("_", "-") log.debug1("%s.addModule('%s')", self._log_prefix, module) self.parent.accessCheck(sender) settings = list(self.getSettings()) if module in settings[4]: raise FirewallError(errors.ALREADY_ENABLED, module) settings[4].append(module) self.update(settings) @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_SERVICE, in_signature='s') @dbus_handle_exceptions def removeModule(self, module, sender=None): module = dbus_to_python(module, str) if module.startswith("nf_conntrack_"): module = module.replace("nf_conntrack_", "") if "_" in module: module = module.replace("_", "-") log.debug1("%s.removeModule('%s')", self._log_prefix, module) self.parent.accessCheck(sender) settings = list(self.getSettings()) if module not in settings[4]: raise FirewallError(errors.NOT_ENABLED, module) settings[4].remove(module) self.update(settings) @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_SERVICE, in_signature='s', out_signature='b') @dbus_handle_exceptions def queryModule(self, module, sender=None): # pylint: disable=W0613 module = dbus_to_python(module, str) if module.startswith("nf_conntrack_"): module = module.replace("nf_conntrack_", "") if "_" in module: module = module.replace("_", "-") log.debug1("%s.queryModule('%s')", self._log_prefix, module) return module in self.getSettings()[4] # destination @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_SERVICE, out_signature='a{ss}') @dbus_handle_exceptions def getDestinations(self, sender=None): # pylint: disable=W0613 log.debug1("%s.getDestinations()", self._log_prefix) return self.getSettings()[5] @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_SERVICE, in_signature='a{ss}') @dbus_handle_exceptions def setDestinations(self, destinations, sender=None): destinations = dbus_to_python(destinations, dict) log.debug1("%s.setDestinations({ipv4:'%s', ipv6:'%s'})", self._log_prefix, destinations.get('ipv4'), destinations.get('ipv6')) self.parent.accessCheck(sender) settings = list(self.getSettings()) settings[5] = destinations self.update(settings) @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_SERVICE, in_signature='s', out_signature='s') @dbus_handle_exceptions def getDestination(self, family, sender=None): family = dbus_to_python(family, str) log.debug1("%s.getDestination('%s')", self._log_prefix, family) self.parent.accessCheck(sender) settings = list(self.getSettings()) if family not in settings[5]: raise FirewallError(errors.NOT_ENABLED, family) return settings[5][family] @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_SERVICE, in_signature='ss') @dbus_handle_exceptions def setDestination(self, family, address, sender=None): family = dbus_to_python(family, str) address = dbus_to_python(address, str) log.debug1("%s.setDestination('%s', '%s')", self._log_prefix, family, address) self.parent.accessCheck(sender) settings = list(self.getSettings()) if family in settings[5] and settings[5][family] == address: raise FirewallError(errors.ALREADY_ENABLED, "'%s': '%s'" % (family, address)) settings[5][family] = address self.update(settings) @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_SERVICE, in_signature='s') @dbus_handle_exceptions def removeDestination(self, family, sender=None): family = dbus_to_python(family, str) log.debug1("%s.removeDestination('%s')", self._log_prefix, family) self.parent.accessCheck(sender) settings = list(self.getSettings()) if family not in settings[5]: raise FirewallError(errors.NOT_ENABLED, family) del settings[5][family] self.update(settings) @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_SERVICE, in_signature='ss', out_signature='b') @dbus_handle_exceptions def queryDestination(self, family, address, sender=None): # pylint: disable=W0613 family = dbus_to_python(family, str) address = dbus_to_python(address, str) log.debug1("%s.queryDestination('%s', '%s')", self._log_prefix, family, address) settings = self.getSettings() return (family in settings[5] and address == settings[5][family]) # includes @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_SERVICE, out_signature='as') @dbus_handle_exceptions def getIncludes(self, sender=None): log.debug1("%s.getIncludes()", self._log_prefix) self.parent.accessCheck(sender) settings = self.config.get_service_config_dict(self.obj) return settings["includes"] if "includes" in settings else [] @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_SERVICE, in_signature='as') @dbus_handle_exceptions def setIncludes(self, includes, sender=None): includes = dbus_to_python(includes, list) log.debug1("%s.setIncludes('%s')", self._log_prefix, includes) self.parent.accessCheck(sender) settings = {"includes": includes[:]} self.obj = self.config.set_service_config_dict(self.obj, settings) self.Updated(self.obj.name) @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_SERVICE, in_signature='s') @dbus_handle_exceptions def addInclude(self, include, sender=None): include = dbus_to_python(include, str) log.debug1("%s.addInclude('%s')", self._log_prefix, include) self.parent.accessCheck(sender) settings = self.config.get_service_config_dict(self.obj) settings.setdefault("includes", []).append(include) self.obj = self.config.set_service_config_dict(self.obj, settings) self.Updated(self.obj.name) @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_SERVICE, in_signature='s') @dbus_handle_exceptions def removeInclude(self, include, sender=None): include = dbus_to_python(include, str) log.debug1("%s.removeInclude('%s')", self._log_prefix, include) self.parent.accessCheck(sender) settings = self.config.get_service_config_dict(self.obj) settings["includes"].remove(include) self.obj = self.config.set_service_config_dict(self.obj, settings) self.Updated(self.obj.name) @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_SERVICE, in_signature='s', out_signature='b') @dbus_handle_exceptions def queryInclude(self, include, sender=None): include = dbus_to_python(include, str) log.debug1("%s.queryInclude('%s')", self._log_prefix, include) settings = self.config.get_service_config_dict(self.obj) return include in settings["includes"] if "includes" in settings else False	29.34 KB	2026-03-10 23:36
📄 config_zone.py # -- coding: utf-8 -- # # Copyright (C) 2010-2016 Red Hat, Inc. # # Authors: # Thomas Woerner <twoerner@redhat.com> # # This program is free software; you can redistribute it and/or modify # it under the terms of the GNU General Public License as published by # the Free Software Foundation; either version 2 of the License, or # (at your option) any later version. # # This program is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU General Public License for more details. # # You should have received a copy of the GNU General Public License # along with this program. If not, see <http://www.gnu.org/licenses/>. import dbus import dbus.service from firewall import config from firewall.dbus_utils import dbus_to_python, \ dbus_introspection_prepare_properties, \ dbus_introspection_add_properties from firewall.core.io.zone import Zone from firewall.core.fw_ifcfg import ifcfg_set_zone_of_interface from firewall.core.base import DEFAULT_ZONE_TARGET from firewall.core.rich import Rich_Rule from firewall.core.logger import log from firewall.server.dbus import DbusServiceObject from firewall.server.decorators import handle_exceptions, \ dbus_handle_exceptions, dbus_service_method, \ dbus_polkit_require_auth from firewall import errors from firewall.errors import FirewallError from firewall.functions import portStr, portInPortRange, coalescePortRange, \ breakPortRange ############################################################################ # # class FirewallDConfig # ############################################################################ class FirewallDConfigZone(DbusServiceObject): """FirewallD main class""" persistent = True """ Make FirewallD persistent. """ default_polkit_auth_required = config.dbus.PK_ACTION_CONFIG """ Use PK_ACTION_INFO as a default """ @handle_exceptions def __init__(self, parent, conf, zone, item_id, args, kwargs): super(FirewallDConfigZone, self).__init__(args, **kwargs) self.parent = parent self.config = conf self.obj = zone self.item_id = item_id self.busname = args[0] self.path = args[1] self._log_prefix = "config.zone.%d" % self.item_id dbus_introspection_prepare_properties( self, config.dbus.DBUS_INTERFACE_CONFIG_ZONE) @dbus_handle_exceptions def __del__(self): pass @dbus_handle_exceptions def unregister(self): self.remove_from_connection() # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # P R O P E R T I E S @dbus_handle_exceptions def _get_property(self, property_name): if property_name == "name": return dbus.String(self.obj.name) elif property_name == "filename": return dbus.String(self.obj.filename) elif property_name == "path": return dbus.String(self.obj.path) elif property_name == "default": return dbus.Boolean(self.obj.default) elif property_name == "builtin": return dbus.Boolean(self.obj.builtin) else: raise dbus.exceptions.DBusException( "org.freedesktop.DBus.Error.InvalidArgs: " "Property '%s' does not exist" % property_name) @dbus_service_method(dbus.PROPERTIES_IFACE, in_signature='ss', out_signature='v') @dbus_handle_exceptions def Get(self, interface_name, property_name, sender=None): # pylint: disable=W0613 # get a property interface_name = dbus_to_python(interface_name, str) property_name = dbus_to_python(property_name, str) log.debug1("%s.Get('%s', '%s')", self._log_prefix, interface_name, property_name) if interface_name != config.dbus.DBUS_INTERFACE_CONFIG_ZONE: raise dbus.exceptions.DBusException( "org.freedesktop.DBus.Error.UnknownInterface: " "Interface '%s' does not exist" % interface_name) return self._get_property(property_name) @dbus_service_method(dbus.PROPERTIES_IFACE, in_signature='s', out_signature='a{sv}') @dbus_handle_exceptions def GetAll(self, interface_name, sender=None): # pylint: disable=W0613 interface_name = dbus_to_python(interface_name, str) log.debug1("%s.GetAll('%s')", self._log_prefix, interface_name) if interface_name != config.dbus.DBUS_INTERFACE_CONFIG_ZONE: raise dbus.exceptions.DBusException( "org.freedesktop.DBus.Error.UnknownInterface: " "Interface '%s' does not exist" % interface_name) ret = { } for x in [ "name", "filename", "path", "default", "builtin" ]: ret[x] = self._get_property(x) return dbus.Dictionary(ret, signature="sv") @dbus_polkit_require_auth(config.dbus.PK_ACTION_CONFIG) @dbus_service_method(dbus.PROPERTIES_IFACE, in_signature='ssv') @dbus_handle_exceptions def Set(self, interface_name, property_name, new_value, sender=None): interface_name = dbus_to_python(interface_name, str) property_name = dbus_to_python(property_name, str) new_value = dbus_to_python(new_value) log.debug1("%s.Set('%s', '%s', '%s')", self._log_prefix, interface_name, property_name, new_value) self.parent.accessCheck(sender) if interface_name != config.dbus.DBUS_INTERFACE_CONFIG_ZONE: raise dbus.exceptions.DBusException( "org.freedesktop.DBus.Error.UnknownInterface: " "Interface '%s' does not exist" % interface_name) raise dbus.exceptions.DBusException( "org.freedesktop.DBus.Error.PropertyReadOnly: " "Property '%s' is read-only" % property_name) @dbus.service.signal(dbus.PROPERTIES_IFACE, signature='sa{sv}as') def PropertiesChanged(self, interface_name, changed_properties, invalidated_properties): interface_name = dbus_to_python(interface_name, str) changed_properties = dbus_to_python(changed_properties) invalidated_properties = dbus_to_python(invalidated_properties) log.debug1("%s.PropertiesChanged('%s', '%s', '%s')", self._log_prefix, interface_name, changed_properties, invalidated_properties) @dbus_polkit_require_auth(config.dbus.PK_ACTION_INFO) @dbus_service_method(dbus.INTROSPECTABLE_IFACE, out_signature='s') @dbus_handle_exceptions def Introspect(self, sender=None): # pylint: disable=W0613 log.debug2("%s.Introspect()", self._log_prefix) data = super(FirewallDConfigZone, self).Introspect( self.path, self.busname.get_bus()) return dbus_introspection_add_properties( self, data, config.dbus.DBUS_INTERFACE_CONFIG_ZONE) # S E T T I N G S @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_ZONE, out_signature="(sssbsasa(ss)asba(ssss)asasasasa(ss)b)") @dbus_handle_exceptions def getSettings(self, sender=None): # pylint: disable=W0613 """get settings for zone """ log.debug1("%s.getSettings()", self._log_prefix) settings = self.config.get_zone_config(self.obj) if settings[4] == DEFAULT_ZONE_TARGET: # convert to list, fix target, convert back to tuple _settings = list(settings) _settings[4] = "default" settings = tuple(_settings) return settings @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_ZONE, out_signature="a{sv}") @dbus_handle_exceptions def getSettings2(self, sender=None): """get settings for zone """ log.debug1("%s.getSettings2()", self._log_prefix) settings = self.config.get_zone_config_dict(self.obj) if settings["target"] == DEFAULT_ZONE_TARGET: settings["target"] = "default" return settings def _checkDuplicateInterfacesSources(self, settings): """Assignment of interfaces/sources to zones is different from other zone settings in the sense that particular interface/zone can be part of only one zone. So make sure added interfaces/sources have not already been bound to another zone.""" old_settings = self.config.get_zone_config_dict(self.obj) old_ifaces = set(old_settings["interfaces"]) if "interfaces" in old_settings else set() old_sources = set(old_settings["sources"]) if "sources" in old_settings else set() if isinstance(settings, tuple): added_ifaces = set(settings[Zone.index_of("interfaces")]) - old_ifaces added_sources = set(settings[Zone.index_of("sources")]) - old_sources else: # dict new_ifaces = set(settings["interfaces"]) if "interfaces" in settings else set() new_sources = set(settings["sources"]) if "sources" in settings else set() added_ifaces = new_ifaces - old_ifaces added_sources = new_sources - old_sources for iface in added_ifaces: if self.parent.getZoneOfInterface(iface): raise FirewallError(errors.ZONE_CONFLICT, iface) # or move to new zone ? for source in added_sources: if self.parent.getZoneOfSource(source): raise FirewallError(errors.ZONE_CONFLICT, source) # or move to new zone ? @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_ZONE, in_signature="(sssbsasa(ss)asba(ssss)asasasasa(ss)b)") @dbus_handle_exceptions def update(self, settings, sender=None): """update settings for zone """ settings = dbus_to_python(settings) log.debug1("%s.update('...')", self._log_prefix) self.parent.accessCheck(sender) if settings[4] == "default": # convert to list, fix target, convert back to tuple _settings = list(settings) _settings[4] = DEFAULT_ZONE_TARGET settings = tuple(_settings) self._checkDuplicateInterfacesSources(settings) self.obj = self.config.set_zone_config(self.obj, settings) self.Updated(self.obj.name) @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_ZONE, in_signature="a{sv}") @dbus_handle_exceptions def update2(self, settings, sender=None): """update settings for zone """ settings = dbus_to_python(settings) log.debug1("%s.update2('...')", self._log_prefix) self.parent.accessCheck(sender) if "target" in settings and settings["target"] == "default": settings["target"] = DEFAULT_ZONE_TARGET self._checkDuplicateInterfacesSources(settings) self.obj = self.config.set_zone_config_dict(self.obj, settings) self.Updated(self.obj.name) @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_ZONE) @dbus_handle_exceptions def loadDefaults(self, sender=None): """load default settings for builtin zone """ log.debug1("%s.loadDefaults()", self._log_prefix) self.parent.accessCheck(sender) self.obj = self.config.load_zone_defaults(self.obj) self.Updated(self.obj.name) @dbus.service.signal(config.dbus.DBUS_INTERFACE_CONFIG_ZONE, signature='s') @dbus_handle_exceptions def Updated(self, name): log.debug1("%s.Updated('%s')" % (self._log_prefix, name)) # R E M O V E @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_ZONE) @dbus_handle_exceptions def remove(self, sender=None): """remove zone """ log.debug1("%s.removeZone()", self._log_prefix) self.parent.accessCheck(sender) self.config.remove_zone(self.obj) self.parent.removeZone(self.obj) @dbus.service.signal(config.dbus.DBUS_INTERFACE_CONFIG_ZONE, signature='s') @dbus_handle_exceptions def Removed(self, name): log.debug1("%s.Removed('%s')" % (self._log_prefix, name)) # R E N A M E @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_ZONE, in_signature='s') @dbus_handle_exceptions def rename(self, name, sender=None): """rename zone """ name = dbus_to_python(name, str) log.debug1("%s.rename('%s')", self._log_prefix, name) self.parent.accessCheck(sender) self.obj = self.config.rename_zone(self.obj, name) self.Renamed(name) @dbus.service.signal(config.dbus.DBUS_INTERFACE_CONFIG_ZONE, signature='s') @dbus_handle_exceptions def Renamed(self, name): log.debug1("%s.Renamed('%s')" % (self._log_prefix, name)) # version @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_ZONE, out_signature='s') @dbus_handle_exceptions def getVersion(self, sender=None): # pylint: disable=W0613 log.debug1("%s.getVersion()", self._log_prefix) return self.getSettings()[0] @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_ZONE, in_signature='s') @dbus_handle_exceptions def setVersion(self, version, sender=None): version = dbus_to_python(version, str) log.debug1("%s.setVersion('%s')", self._log_prefix, version) self.parent.accessCheck(sender) settings = list(self.getSettings()) settings[0] = version self.update(settings) # short @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_ZONE, out_signature='s') @dbus_handle_exceptions def getShort(self, sender=None): # pylint: disable=W0613 log.debug1("%s.getShort()", self._log_prefix) return self.getSettings()[1] @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_ZONE, in_signature='s') @dbus_handle_exceptions def setShort(self, short, sender=None): short = dbus_to_python(short, str) log.debug1("%s.setShort('%s')", self._log_prefix, short) self.parent.accessCheck(sender) settings = list(self.getSettings()) settings[1] = short self.update(settings) # description @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_ZONE, out_signature='s') @dbus_handle_exceptions def getDescription(self, sender=None): # pylint: disable=W0613 log.debug1("%s.getDescription()", self._log_prefix) return self.getSettings()[2] @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_ZONE, in_signature='s') @dbus_handle_exceptions def setDescription(self, description, sender=None): description = dbus_to_python(description, str) log.debug1("%s.setDescription('%s')", self._log_prefix, description) self.parent.accessCheck(sender) settings = list(self.getSettings()) settings[2] = description self.update(settings) # immutable (deprecated) # settings[3] was used for 'immutable' # target @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_ZONE, out_signature='s') @dbus_handle_exceptions def getTarget(self, sender=None): # pylint: disable=W0613 log.debug1("%s.getTarget()", self._log_prefix) settings = self.getSettings() return settings[4] if settings[4] != DEFAULT_ZONE_TARGET else "default" @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_ZONE, in_signature='s') @dbus_handle_exceptions def setTarget(self, target, sender=None): target = dbus_to_python(target, str) log.debug1("%s.setTarget('%s')", self._log_prefix, target) self.parent.accessCheck(sender) settings = list(self.getSettings()) settings[4] = target if target != "default" else DEFAULT_ZONE_TARGET self.update(settings) # service @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_ZONE, out_signature='as') @dbus_handle_exceptions def getServices(self, sender=None): # pylint: disable=W0613 log.debug1("%s.getServices()", self._log_prefix) return self.getSettings()[5] @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_ZONE, in_signature='as') @dbus_handle_exceptions def setServices(self, services, sender=None): services = dbus_to_python(services, list) log.debug1("%s.setServices('[%s]')", self._log_prefix, ",".join(services)) self.parent.accessCheck(sender) settings = list(self.getSettings()) settings[5] = services self.update(settings) @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_ZONE, in_signature='s') @dbus_handle_exceptions def addService(self, service, sender=None): service = dbus_to_python(service, str) log.debug1("%s.addService('%s')", self._log_prefix, service) self.parent.accessCheck(sender) settings = list(self.getSettings()) if service in settings[5]: raise FirewallError(errors.ALREADY_ENABLED, service) settings[5].append(service) self.update(settings) @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_ZONE, in_signature='s') @dbus_handle_exceptions def removeService(self, service, sender=None): service = dbus_to_python(service, str) log.debug1("%s.removeService('%s')", self._log_prefix, service) self.parent.accessCheck(sender) settings = list(self.getSettings()) if service not in settings[5]: raise FirewallError(errors.NOT_ENABLED, service) settings[5].remove(service) self.update(settings) @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_ZONE, in_signature='s', out_signature='b') @dbus_handle_exceptions def queryService(self, service, sender=None): # pylint: disable=W0613 service = dbus_to_python(service, str) log.debug1("%s.queryService('%s')", self._log_prefix, service) return service in self.getSettings()[5] # port @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_ZONE, out_signature='a(ss)') @dbus_handle_exceptions def getPorts(self, sender=None): # pylint: disable=W0613 log.debug1("%s.getPorts()", self._log_prefix) return self.getSettings()[6] @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_ZONE, in_signature='a(ss)') @dbus_handle_exceptions def setPorts(self, ports, sender=None): _ports = [ ] # convert embedded lists to tuples for port in dbus_to_python(ports, list): if isinstance(port, list): _ports.append(tuple(port)) else: _ports.append(port) ports = _ports log.debug1("%s.setPorts('[%s]')", self._log_prefix, ",".join("('%s, '%s')" % (port[0], port[1]) for port in ports)) self.parent.accessCheck(sender) settings = list(self.getSettings()) settings[6] = ports self.update(settings) @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_ZONE, in_signature='ss') @dbus_handle_exceptions def addPort(self, port, protocol, sender=None): port = dbus_to_python(port, str) protocol = dbus_to_python(protocol, str) log.debug1("%s.addPort('%s', '%s')", self._log_prefix, port, protocol) self.parent.accessCheck(sender) settings = list(self.getSettings()) existing_port_ids = list(filter(lambda x: x[1] == protocol, settings[6])) for port_id in existing_port_ids: if portInPortRange(port, port_id[0]): raise FirewallError(errors.ALREADY_ENABLED, "%s:%s" % (port, protocol)) added_ranges, removed_ranges = coalescePortRange(port, [_port for (_port, _protocol) in existing_port_ids]) for range in removed_ranges: settings[6].remove((portStr(range, "-"), protocol)) for range in added_ranges: settings[6].append((portStr(range, "-"), protocol)) self.update(settings) @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_ZONE, in_signature='ss') @dbus_handle_exceptions def removePort(self, port, protocol, sender=None): port = dbus_to_python(port, str) protocol = dbus_to_python(protocol, str) log.debug1("%s.removePort('%s', '%s')", self._log_prefix, port, protocol) self.parent.accessCheck(sender) settings = list(self.getSettings()) existing_port_ids = list(filter(lambda x: x[1] == protocol, settings[6])) for port_id in existing_port_ids: if portInPortRange(port, port_id[0]): break else: raise FirewallError(errors.NOT_ENABLED, "%s:%s" % (port, protocol)) added_ranges, removed_ranges = breakPortRange(port, [_port for (_port, _protocol) in existing_port_ids]) for range in removed_ranges: settings[6].remove((portStr(range, "-"), protocol)) for range in added_ranges: settings[6].append((portStr(range, "-"), protocol)) self.update(settings) @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_ZONE, in_signature='ss', out_signature='b') @dbus_handle_exceptions def queryPort(self, port, protocol, sender=None): # pylint: disable=W0613 port = dbus_to_python(port, str) protocol = dbus_to_python(protocol, str) log.debug1("%s.queryPort('%s', '%s')", self._log_prefix, port, protocol) for (_port, _protocol) in self.getSettings()[6]: if portInPortRange(port, _port) and protocol == _protocol: return True return False # protocol @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_ZONE, out_signature='as') @dbus_handle_exceptions def getProtocols(self, sender=None): # pylint: disable=W0613 log.debug1("%s.getProtocols()", self._log_prefix) return self.getSettings()[13] @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_ZONE, in_signature='as') @dbus_handle_exceptions def setProtocols(self, protocols, sender=None): protocols = dbus_to_python(protocols, list) log.debug1("%s.setProtocols('[%s]')", self._log_prefix, ",".join(protocols)) self.parent.accessCheck(sender) settings = list(self.getSettings()) settings[13] = protocols self.update(settings) @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_ZONE, in_signature='s') @dbus_handle_exceptions def addProtocol(self, protocol, sender=None): protocol = dbus_to_python(protocol, str) log.debug1("%s.addProtocol('%s')", self._log_prefix, protocol) self.parent.accessCheck(sender) settings = list(self.getSettings()) if protocol in settings[13]: raise FirewallError(errors.ALREADY_ENABLED, protocol) settings[13].append(protocol) self.update(settings) @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_ZONE, in_signature='s') @dbus_handle_exceptions def removeProtocol(self, protocol, sender=None): protocol = dbus_to_python(protocol, str) log.debug1("%s.removeProtocol('%s')", self._log_prefix, protocol) self.parent.accessCheck(sender) settings = list(self.getSettings()) if protocol not in settings[13]: raise FirewallError(errors.NOT_ENABLED, protocol) settings[13].remove(protocol) self.update(settings) @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_ZONE, in_signature='s', out_signature='b') @dbus_handle_exceptions def queryProtocol(self, protocol, sender=None): # pylint: disable=W0613 protocol = dbus_to_python(protocol, str) log.debug1("%s.queryProtocol('%s')", self._log_prefix, protocol) return protocol in self.getSettings()[13] # source port @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_ZONE, out_signature='a(ss)') @dbus_handle_exceptions def getSourcePorts(self, sender=None): # pylint: disable=W0613 log.debug1("%s.getSourcePorts()", self._log_prefix) return self.getSettings()[14] @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_ZONE, in_signature='a(ss)') @dbus_handle_exceptions def setSourcePorts(self, ports, sender=None): _ports = [ ] # convert embedded lists to tuples for port in dbus_to_python(ports, list): if isinstance(port, list): _ports.append(tuple(port)) else: _ports.append(port) ports = _ports log.debug1("%s.setSourcePorts('[%s]')", self._log_prefix, ",".join("('%s, '%s')" % (port[0], port[1]) for port in ports)) self.parent.accessCheck(sender) settings = list(self.getSettings()) settings[14] = ports self.update(settings) @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_ZONE, in_signature='ss') @dbus_handle_exceptions def addSourcePort(self, port, protocol, sender=None): port = dbus_to_python(port, str) protocol = dbus_to_python(protocol, str) log.debug1("%s.addSourcePort('%s', '%s')", self._log_prefix, port, protocol) self.parent.accessCheck(sender) settings = list(self.getSettings()) existing_port_ids = list(filter(lambda x: x[1] == protocol, settings[14])) for port_id in existing_port_ids: if portInPortRange(port, port_id[0]): raise FirewallError(errors.ALREADY_ENABLED, "%s:%s" % (port, protocol)) added_ranges, removed_ranges = coalescePortRange(port, [_port for (_port, _protocol) in existing_port_ids]) for range in removed_ranges: settings[14].remove((portStr(range, "-"), protocol)) for range in added_ranges: settings[14].append((portStr(range, "-"), protocol)) self.update(settings) @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_ZONE, in_signature='ss') @dbus_handle_exceptions def removeSourcePort(self, port, protocol, sender=None): port = dbus_to_python(port, str) protocol = dbus_to_python(protocol, str) log.debug1("%s.removeSourcePort('%s', '%s')", self._log_prefix, port, protocol) self.parent.accessCheck(sender) settings = list(self.getSettings()) existing_port_ids = list(filter(lambda x: x[1] == protocol, settings[14])) for port_id in existing_port_ids: if portInPortRange(port, port_id[0]): break else: raise FirewallError(errors.NOT_ENABLED, "%s:%s" % (port, protocol)) added_ranges, removed_ranges = breakPortRange(port, [_port for (_port, _protocol) in existing_port_ids]) for range in removed_ranges: settings[14].remove((portStr(range, "-"), protocol)) for range in added_ranges: settings[14].append((portStr(range, "-"), protocol)) self.update(settings) @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_ZONE, in_signature='ss', out_signature='b') @dbus_handle_exceptions def querySourcePort(self, port, protocol, sender=None): # pylint: disable=W0613 port = dbus_to_python(port, str) protocol = dbus_to_python(protocol, str) log.debug1("%s.querySourcePort('%s', '%s')", self._log_prefix, port, protocol) for (_port, _protocol) in self.getSettings()[14]: if portInPortRange(port, _port) and protocol == _protocol: return True return False # icmp block @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_ZONE, out_signature='as') @dbus_handle_exceptions def getIcmpBlocks(self, sender=None): # pylint: disable=W0613 log.debug1("%s.getIcmpBlocks()", self._log_prefix) return self.getSettings()[7] @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_ZONE, in_signature='as') @dbus_handle_exceptions def setIcmpBlocks(self, icmptypes, sender=None): icmptypes = dbus_to_python(icmptypes, list) log.debug1("%s.setIcmpBlocks('[%s]')", self._log_prefix, ",".join(icmptypes)) self.parent.accessCheck(sender) settings = list(self.getSettings()) settings[7] = icmptypes self.update(settings) @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_ZONE, in_signature='s') @dbus_handle_exceptions def addIcmpBlock(self, icmptype, sender=None): icmptype = dbus_to_python(icmptype, str) log.debug1("%s.addIcmpBlock('%s')", self._log_prefix, icmptype) self.parent.accessCheck(sender) settings = list(self.getSettings()) if icmptype in settings[7]: raise FirewallError(errors.ALREADY_ENABLED, icmptype) settings[7].append(icmptype) self.update(settings) @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_ZONE, in_signature='s') @dbus_handle_exceptions def removeIcmpBlock(self, icmptype, sender=None): icmptype = dbus_to_python(icmptype, str) log.debug1("%s.removeIcmpBlock('%s')", self._log_prefix, icmptype) self.parent.accessCheck(sender) settings = list(self.getSettings()) if icmptype not in settings[7]: raise FirewallError(errors.NOT_ENABLED, icmptype) settings[7].remove(icmptype) self.update(settings) @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_ZONE, in_signature='s', out_signature='b') @dbus_handle_exceptions def queryIcmpBlock(self, icmptype, sender=None): # pylint: disable=W0613 icmptype = dbus_to_python(icmptype, str) log.debug1("%s.queryIcmpBlock('%s')", self._log_prefix, icmptype) return icmptype in self.getSettings()[7] # icmp block inversion @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_ZONE, out_signature='b') @dbus_handle_exceptions def getIcmpBlockInversion(self, sender=None): # pylint: disable=W0613 log.debug1("%s.getIcmpBlockInversion()", self._log_prefix) return self.getSettings()[15] @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_ZONE, in_signature='b') @dbus_handle_exceptions def setIcmpBlockInversion(self, flag, sender=None): flag = dbus_to_python(flag, bool) log.debug1("%s.setIcmpBlockInversion('%s')", self._log_prefix, flag) self.parent.accessCheck(sender) settings = list(self.getSettings()) settings[15] = flag self.update(settings) @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_ZONE) @dbus_handle_exceptions def addIcmpBlockInversion(self, sender=None): log.debug1("%s.addIcmpBlockInversion()", self._log_prefix) self.parent.accessCheck(sender) settings = list(self.getSettings()) if settings[15]: raise FirewallError(errors.ALREADY_ENABLED, "icmp-block-inversion") settings[15] = True self.update(settings) @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_ZONE) @dbus_handle_exceptions def removeIcmpBlockInversion(self, sender=None): log.debug1("%s.removeIcmpBlockInversion()", self._log_prefix) self.parent.accessCheck(sender) settings = list(self.getSettings()) if not settings[15]: raise FirewallError(errors.NOT_ENABLED, "icmp-block-inversion") settings[15] = False self.update(settings) @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_ZONE, out_signature='b') @dbus_handle_exceptions def queryIcmpBlockInversion(self, sender=None): # pylint: disable=W0613 log.debug1("%s.queryIcmpBlockInversion()", self._log_prefix) return self.getSettings()[15] # masquerade @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_ZONE, out_signature='b') @dbus_handle_exceptions def getMasquerade(self, sender=None): # pylint: disable=W0613 log.debug1("%s.getMasquerade()", self._log_prefix) return self.getSettings()[8] @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_ZONE, in_signature='b') @dbus_handle_exceptions def setMasquerade(self, masquerade, sender=None): masquerade = dbus_to_python(masquerade, bool) log.debug1("%s.setMasquerade('%s')", self._log_prefix, masquerade) self.parent.accessCheck(sender) settings = list(self.getSettings()) settings[8] = masquerade self.update(settings) @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_ZONE) @dbus_handle_exceptions def addMasquerade(self, sender=None): log.debug1("%s.addMasquerade()", self._log_prefix) self.parent.accessCheck(sender) settings = list(self.getSettings()) if settings[8]: raise FirewallError(errors.ALREADY_ENABLED, "masquerade") settings[8] = True self.update(settings) @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_ZONE) @dbus_handle_exceptions def removeMasquerade(self, sender=None): log.debug1("%s.removeMasquerade()", self._log_prefix) self.parent.accessCheck(sender) settings = list(self.getSettings()) if not settings[8]: raise FirewallError(errors.NOT_ENABLED, "masquerade") settings[8] = False self.update(settings) @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_ZONE, out_signature='b') @dbus_handle_exceptions def queryMasquerade(self, sender=None): # pylint: disable=W0613 log.debug1("%s.queryMasquerade()", self._log_prefix) return self.getSettings()[8] # forward port @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_ZONE, out_signature='a(ssss)') @dbus_handle_exceptions def getForwardPorts(self, sender=None): # pylint: disable=W0613 log.debug1("%s.getForwardPorts()", self._log_prefix) return self.getSettings()[9] @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_ZONE, in_signature='a(ssss)') @dbus_handle_exceptions def setForwardPorts(self, ports, sender=None): _ports = [ ] # convert embedded lists to tuples for port in dbus_to_python(ports, list): if isinstance(port, list): _ports.append(tuple(port)) else: _ports.append(port) ports = _ports log.debug1("%s.setForwardPorts('[%s]')", self._log_prefix, ",".join("('%s, '%s', '%s', '%s')" % (port[0], port[1], \ port[2], port[3]) for port in ports)) self.parent.accessCheck(sender) settings = list(self.getSettings()) settings[9] = ports self.update(settings) @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_ZONE, in_signature='ssss') @dbus_handle_exceptions def addForwardPort(self, port, protocol, toport, toaddr, sender=None): # pylint: disable=R0913 port = dbus_to_python(port, str) protocol = dbus_to_python(protocol, str) toport = dbus_to_python(toport, str) toaddr = dbus_to_python(toaddr, str) log.debug1("%s.addForwardPort('%s', '%s', '%s', '%s')", self._log_prefix, port, protocol, toport, toaddr) self.parent.accessCheck(sender) fwp_id = (port, protocol, str(toport), str(toaddr)) settings = list(self.getSettings()) if fwp_id in settings[9]: raise FirewallError(errors.ALREADY_ENABLED, "%s:%s:%s:%s" % (port, protocol, toport, toaddr)) settings[9].append(fwp_id) self.update(settings) @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_ZONE, in_signature='ssss') @dbus_handle_exceptions def removeForwardPort(self, port, protocol, toport, toaddr, sender=None): # pylint: disable=R0913 port = dbus_to_python(port, str) protocol = dbus_to_python(protocol, str) toport = dbus_to_python(toport, str) toaddr = dbus_to_python(toaddr, str) log.debug1("%s.removeForwardPort('%s', '%s', '%s', '%s')", self._log_prefix, port, protocol, toport, toaddr) self.parent.accessCheck(sender) fwp_id = (port, protocol, str(toport), str(toaddr)) settings = list(self.getSettings()) if fwp_id not in settings[9]: raise FirewallError(errors.NOT_ENABLED, "%s:%s:%s:%s" % (port, protocol, toport, toaddr)) settings[9].remove(fwp_id) self.update(settings) @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_ZONE, in_signature='ssss', out_signature='b') @dbus_handle_exceptions def queryForwardPort(self, port, protocol, toport, toaddr, sender=None): # pylint: disable=W0613, R0913 port = dbus_to_python(port, str) protocol = dbus_to_python(protocol, str) toport = dbus_to_python(toport, str) toaddr = dbus_to_python(toaddr, str) log.debug1("%s.queryForwardPort('%s', '%s', '%s', '%s')", self._log_prefix, port, protocol, toport, toaddr) fwp_id = (port, protocol, str(toport), str(toaddr)) return fwp_id in self.getSettings()[9] # interface @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_ZONE, out_signature='as') @dbus_handle_exceptions def getInterfaces(self, sender=None): # pylint: disable=W0613 log.debug1("%s.getInterfaces()", self._log_prefix) return self.getSettings()[10] @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_ZONE, in_signature='as') @dbus_handle_exceptions def setInterfaces(self, interfaces, sender=None): interfaces = dbus_to_python(interfaces, list) log.debug1("%s.setInterfaces('[%s]')", self._log_prefix, ",".join(interfaces)) self.parent.accessCheck(sender) settings = list(self.getSettings()) settings[10] = interfaces self.update(settings) @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_ZONE, in_signature='s') @dbus_handle_exceptions def addInterface(self, interface, sender=None): interface = dbus_to_python(interface, str) log.debug1("%s.addInterface('%s')", self._log_prefix, interface) self.parent.accessCheck(sender) settings = list(self.getSettings()) if interface in settings[10]: raise FirewallError(errors.ALREADY_ENABLED, interface) settings[10].append(interface) self.update(settings) ifcfg_set_zone_of_interface(self.obj.name, interface) @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_ZONE, in_signature='s') @dbus_handle_exceptions def removeInterface(self, interface, sender=None): interface = dbus_to_python(interface, str) log.debug1("%s.removeInterface('%s')", self._log_prefix, interface) self.parent.accessCheck(sender) settings = list(self.getSettings()) if interface not in settings[10]: raise FirewallError(errors.NOT_ENABLED, interface) settings[10].remove(interface) self.update(settings) ifcfg_set_zone_of_interface("", interface) @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_ZONE, in_signature='s', out_signature='b') @dbus_handle_exceptions def queryInterface(self, interface, sender=None): # pylint: disable=W0613 interface = dbus_to_python(interface, str) log.debug1("%s.queryInterface('%s')", self._log_prefix, interface) return interface in self.getSettings()[10] # source @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_ZONE, out_signature='as') @dbus_handle_exceptions def getSources(self, sender=None): # pylint: disable=W0613 log.debug1("%s.getSources()", self._log_prefix) return self.getSettings()[11] @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_ZONE, in_signature='as') @dbus_handle_exceptions def setSources(self, sources, sender=None): sources = dbus_to_python(sources, list) log.debug1("%s.setSources('[%s]')", self._log_prefix, ",".join(sources)) self.parent.accessCheck(sender) settings = list(self.getSettings()) settings[11] = sources self.update(settings) @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_ZONE, in_signature='s') @dbus_handle_exceptions def addSource(self, source, sender=None): source = dbus_to_python(source, str) log.debug1("%s.addSource('%s')", self._log_prefix, source) self.parent.accessCheck(sender) settings = list(self.getSettings()) if source in settings[11]: raise FirewallError(errors.ALREADY_ENABLED, source) settings[11].append(source) self.update(settings) @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_ZONE, in_signature='s') @dbus_handle_exceptions def removeSource(self, source, sender=None): source = dbus_to_python(source, str) log.debug1("%s.removeSource('%s')", self._log_prefix, source) self.parent.accessCheck(sender) settings = list(self.getSettings()) if source not in settings[11]: raise FirewallError(errors.NOT_ENABLED, source) settings[11].remove(source) self.update(settings) @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_ZONE, in_signature='s', out_signature='b') @dbus_handle_exceptions def querySource(self, source, sender=None): # pylint: disable=W0613 source = dbus_to_python(source, str) log.debug1("%s.querySource('%s')", self._log_prefix, source) return source in self.getSettings()[11] # rich rule @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_ZONE, out_signature='as') @dbus_handle_exceptions def getRichRules(self, sender=None): # pylint: disable=W0613 log.debug1("%s.getRichRules()", self._log_prefix) return self.getSettings()[12] @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_ZONE, in_signature='as') @dbus_handle_exceptions def setRichRules(self, rules, sender=None): rules = dbus_to_python(rules, list) log.debug1("%s.setRichRules('[%s]')", self._log_prefix, ",".join(rules)) self.parent.accessCheck(sender) settings = list(self.getSettings()) rules = [ str(Rich_Rule(rule_str=r)) for r in rules ] settings[12] = rules self.update(settings) @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_ZONE, in_signature='s') @dbus_handle_exceptions def addRichRule(self, rule, sender=None): rule = dbus_to_python(rule, str) log.debug1("%s.addRichRule('%s')", self._log_prefix, rule) self.parent.accessCheck(sender) settings = list(self.getSettings()) rule_str = str(Rich_Rule(rule_str=rule)) if rule_str in settings[12]: raise FirewallError(errors.ALREADY_ENABLED, rule) settings[12].append(rule_str) self.update(settings) @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_ZONE, in_signature='s') @dbus_handle_exceptions def removeRichRule(self, rule, sender=None): rule = dbus_to_python(rule, str) log.debug1("%s.removeRichRule('%s')", self._log_prefix, rule) self.parent.accessCheck(sender) settings = list(self.getSettings()) rule_str = str(Rich_Rule(rule_str=rule)) if rule_str not in settings[12]: raise FirewallError(errors.NOT_ENABLED, rule) settings[12].remove(rule_str) self.update(settings) @dbus_service_method(config.dbus.DBUS_INTERFACE_CONFIG_ZONE, in_signature='s', out_signature='b') @dbus_handle_exceptions def queryRichRule(self, rule, sender=None): # pylint: disable=W0613 rule = dbus_to_python(rule, str) log.debug1("%s.queryRichRule('%s')", self._log_prefix, rule) rule_str = str(Rich_Rule(rule_str=rule)) return rule_str in self.getSettings()[12]	44.99 KB	2026-03-10 23:36
📄 dbus.py # -- coding: utf-8 -- # # SPDX-License-Identifier: GPL-2.0-or-later import dbus from firewall import config class FirewallDBusException(dbus.DBusException): """FirewallDBusException""" _dbus_error_name = "%s.Exception" % config.dbus.DBUS_INTERFACE class NotAuthorizedException(dbus.DBusException): def __init__(self, action_id, method, args, kwargs): self._dbus_error_name = config.dbus.DBUS_INTERFACE + ".NotAuthorizedException" super().__init__("Not Authorized({}): {}".format(method, action_id)) class DbusServiceObject(dbus.service.Object): def __new__(cls, args, **kwargs): # Check each dbus method. If it does not have an explicit polkit auth # then implicitly wrap it with the default from firewall.server.decorators import dbus_polkit_require_auth for attr_name in dir(cls): method = getattr(cls, attr_name) if hasattr(method, "_dbus_is_method") and \ not hasattr(method, "_polkit_auth_required"): _decorator = dbus_polkit_require_auth(cls.default_polkit_auth_required) setattr(cls, attr_name, _decorator(method)) return super().__new__(cls)	1.18 KB	2026-03-10 23:36
📄 decorators.py # -- coding: utf-8 -- # # Copyright (C) 2012-2016 Red Hat, Inc. # # Authors: # Thomas Woerner <twoerner@redhat.com> # # This program is free software; you can redistribute it and/or modify # it under the terms of the GNU General Public License as published by # the Free Software Foundation; either version 2 of the License, or # (at your option) any later version. # # This program is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU General Public License for more details. # # You should have received a copy of the GNU General Public License # along with this program. If not, see <http://www.gnu.org/licenses/>. """This module contains decorators for use with and without D-Bus""" __all__ = ["handle_exceptions", "dbus_handle_exceptions", "dbus_service_method"] import dbus import dbus.service import traceback import functools import inspect from dbus.exceptions import DBusException from firewall.errors import FirewallError from firewall import errors from firewall.core.logger import log from firewall.server.dbus import FirewallDBusException, NotAuthorizedException from firewall.dbus_utils import uid_of_sender ############################################################################ # # Exception handler decorators # ############################################################################ def handle_exceptions(func): """Decorator to handle exceptions and log them. Used if not conneced to D-Bus. """ @functools.wraps(func) def _impl(args, kwargs): try: return func(args, *kwargs) except FirewallError as error: log.debug1(traceback.format_exc()) log.error(error) except Exception: # pylint: disable=W0703 log.exception() return _impl def dbus_handle_exceptions(func): """Decorator to handle exceptions, log and report them into D-Bus :Raises DBusException: on a firewall error code problems. """ @functools.wraps(func) def _impl(args, *kwargs): try: return func(args, *kwargs) except FirewallError as error: code = FirewallError.get_code(str(error)) if code in [ errors.ALREADY_ENABLED, errors.NOT_ENABLED, errors.ZONE_ALREADY_SET, errors.ALREADY_SET ]: log.warning(str(error)) else: log.debug1(traceback.format_exc()) log.error(str(error)) raise FirewallDBusException(str(error)) except DBusException as ex: # only log DBusExceptions once raise ex except Exception as ex: log.exception() raise FirewallDBusException(str(ex)) # HACK: functools.wraps() does not copy the function signature and # dbus-python doesn't support varargs. As such we need to copy the # signature from the function to the newly decorated function otherwise the # decorators in dbus-python will manipulate the arg stack and fail # miserably. # # Note: This can be removed if we ever stop using dbus-python. # # Ref: https://gitlab.freedesktop.org/dbus/dbus-python/-/issues/12 # _impl.__signature__ = inspect.signature(func) return _impl def dbus_service_method(args, *kwargs): """Add sender argument for D-Bus""" kwargs.setdefault("sender_keyword", "sender") return dbus.service.method(args, *kwargs) class dbus_service_method_deprecated: """Decorator that maintains a list of deprecated methods in dbus interfaces. """ deprecated = {} def __init__(self, interface=None): self.interface = interface if self.interface: if self.interface not in self.deprecated: self.deprecated[self.interface] = set() def __call__(self, func): if self.interface: self.deprecated[self.interface].add(func.__name__) @functools.wraps(func) def _impl(args, *kwargs): return func(args, *kwargs) return _impl class dbus_service_signal_deprecated(dbus_service_method_deprecated): """Decorator that maintains a list of deprecated signals in dbus interfaces. """ pass class dbus_polkit_require_auth: """Decorator factory that checks if the interface/method can be used by the sender/user. Assumes wrapped function is a method inside a class derived from DbusServiceObject. """ _polkit_name = "org.freedesktop.PolicyKit1" _polkit_path = "/org/freedesktop/PolicyKit1/Authority" _polkit_interface = "org.freedesktop.PolicyKit1.Authority" _bus = None _bus_signal_receiver = None _interface_polkit = None def __init__(self, polkit_auth_required): self._polkit_auth_required = polkit_auth_required @classmethod def _polkit_name_owner_changed(cls, name, old_owner, new_owner): cls._bus.remove_signal_receiver(cls._bus_signal_receiver) cls._bus_signal_receiver = None cls._interface_polkit = None def __call__(self, func): @functools.wraps(func) def _impl(args, *kwargs): if not type(self)._bus: type(self)._bus = dbus.SystemBus() if not type(self)._bus_signal_receiver: type(self)._bus_signal_receiver = type(self)._bus.add_signal_receiver( handler_function=type(self)._polkit_name_owner_changed, signal_name="NameOwnerChanged", dbus_interface="org.freedesktop.DBus", arg0=self._polkit_name) if not type(self)._interface_polkit: try: type(self)._interface_polkit = dbus.Interface(type(self)._bus.get_object( type(self)._polkit_name, type(self)._polkit_path), type(self)._polkit_interface) except dbus.DBusException: # polkit must not be available pass action_id = self._polkit_auth_required if not action_id: raise dbus.DBusException("Not Authorized: No action_id specified.") sender = kwargs.get("sender") if sender: # use polkit if it's available if type(self)._interface_polkit: (result, _, _) = type(self)._interface_polkit.CheckAuthorization( ("system-bus-name", {"name": sender}), action_id, {}, 1, "") if not result: raise NotAuthorizedException(action_id, "polkit") # fallback to checking UID else: uid = uid_of_sender(type(self)._bus, sender) if uid != 0: raise NotAuthorizedException(action_id, "uid") return func(args, **kwargs) _impl._polkit_auth_required = self._polkit_auth_required return _impl	7.38 KB	2026-03-10 23:36
📄 firewalld.py # -- coding: utf-8 -- # # Copyright (C) 2010-2016 Red Hat, Inc. # # Authors: # Thomas Woerner <twoerner@redhat.com> # # This program is free software; you can redistribute it and/or modify # it under the terms of the GNU General Public License as published by # the Free Software Foundation; either version 2 of the License, or # (at your option) any later version. # # This program is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU General Public License for more details. # # You should have received a copy of the GNU General Public License # along with this program. If not, see <http://www.gnu.org/licenses/>. __all__ = [ "FirewallD" ] from gi.repository import GLib import copy import dbus import dbus.service import dbus.mainloop.glib from firewall import config from firewall.core.fw import Firewall from firewall.core.rich import Rich_Rule from firewall.core.logger import log from firewall.client import FirewallClientZoneSettings from firewall.server.dbus import FirewallDBusException, DbusServiceObject from firewall.server.decorators import dbus_handle_exceptions, \ dbus_service_method, \ handle_exceptions, \ dbus_service_method_deprecated, \ dbus_service_signal_deprecated, \ dbus_polkit_require_auth from firewall.server.config import FirewallDConfig from firewall.dbus_utils import dbus_to_python, \ command_of_sender, context_of_sender, uid_of_sender, user_of_uid, \ dbus_introspection_prepare_properties, \ dbus_introspection_add_properties, \ dbus_introspection_add_deprecated from firewall.core.io.functions import check_on_disk_config from firewall.core.io.ipset import IPSet from firewall.core.io.icmptype import IcmpType from firewall.core.io.helper import Helper from firewall.core.fw_nm import nm_get_connection_of_interface, \ nm_set_zone_of_connection from firewall.core.fw_ifcfg import ifcfg_set_zone_of_interface from firewall import errors from firewall.errors import FirewallError ############################################################################ # # class FirewallD # ############################################################################ class FirewallD(DbusServiceObject): """FirewallD main class""" persistent = True """ Make FirewallD persistent. """ default_polkit_auth_required = config.dbus.PK_ACTION_CONFIG """ Use config.dbus.PK_ACTION_CONFIG as a default """ @handle_exceptions def __init__(self): self.fw = Firewall() self.start() dbus.mainloop.glib.DBusGMainLoop(set_as_default=True) bus = dbus.SystemBus() name = dbus.service.BusName(config.dbus.DBUS_INTERFACE, bus=bus) super(FirewallD, self).__init__(name, config.dbus.DBUS_PATH) self.busname = name self.path = config.dbus.DBUS_PATH dbus_introspection_prepare_properties(self, config.dbus.DBUS_INTERFACE) self.config = FirewallDConfig(self.fw.config, self.busname, config.dbus.DBUS_PATH_CONFIG) def __del__(self): self.stop() @handle_exceptions def start(self): # tests if iptables and ip6tables are usable using test functions # loads default firewall rules for iptables and ip6tables log.debug1("start()") self._timeouts = { } return self.fw.start() @handle_exceptions def stop(self): # stops firewall: unloads firewall modules, flushes chains and tables, # resets policies log.debug1("stop()") return self.fw.stop() # lockdown functions @dbus_handle_exceptions def accessCheck(self, sender): if self.fw.policies.query_lockdown(): if sender is None: log.error("Lockdown not possible, sender not set.") return bus = dbus.SystemBus() context = context_of_sender(bus, sender) if self.fw.policies.access_check("context", context): return uid = uid_of_sender(bus, sender) if self.fw.policies.access_check("uid", uid): return user = user_of_uid(uid) if self.fw.policies.access_check("user", user): return command = command_of_sender(bus, sender) if self.fw.policies.access_check("command", command): return raise FirewallError(errors.ACCESS_DENIED, "lockdown is enabled") # timeout functions @dbus_handle_exceptions def addTimeout(self, zone, x, tag): if zone not in self._timeouts: self._timeouts[zone] = { } self._timeouts[zone][x] = tag @dbus_handle_exceptions def removeTimeout(self, zone, x): if zone in self._timeouts and x in self._timeouts[zone]: GLib.source_remove(self._timeouts[zone][x]) del self._timeouts[zone][x] @dbus_handle_exceptions def cleanup_timeouts(self): # cleanup timeouts for zone in self._timeouts: for x in self._timeouts[zone]: GLib.source_remove(self._timeouts[zone][x]) self._timeouts[zone].clear() self._timeouts.clear() # property handling @dbus_handle_exceptions def _get_property(self, prop): if prop == "version": return dbus.String(config.VERSION) elif prop == "interface_version": return dbus.String("%d.%d" % (config.dbus.DBUS_INTERFACE_VERSION, config.dbus.DBUS_INTERFACE_REVISION)) elif prop == "state": return dbus.String(self.fw.get_state()) elif prop == "IPv4": return dbus.Boolean(self.fw.is_ipv_enabled("ipv4")) elif prop == "IPv4ICMPTypes": return dbus.Array(self.fw.ipv4_supported_icmp_types, "s") elif prop == "IPv6": return dbus.Boolean(self.fw.is_ipv_enabled("ipv6")) elif prop == "IPv6_rpfilter": return dbus.Boolean(False if self.fw._ipv6_rpfilter == "no" else True) elif prop == "IPv6ICMPTypes": return dbus.Array(self.fw.ipv6_supported_icmp_types, "s") elif prop == "BRIDGE": return dbus.Boolean(self.fw.ebtables_enabled) elif prop == "IPSet": return dbus.Boolean(self.fw.ipset_enabled) elif prop == "IPSetTypes": return dbus.Array(self.fw.ipset_supported_types, "s") elif prop == "nf_conntrack_helper_setting": return dbus.Boolean(False) elif prop == "nf_conntrack_helpers": return dbus.Dictionary({}, "sas") elif prop == "nf_nat_helpers": return dbus.Dictionary({}, "sas") else: raise dbus.exceptions.DBusException( "org.freedesktop.DBus.Error.InvalidArgs: " "Property '%s' does not exist" % prop) @dbus_service_method(dbus.PROPERTIES_IFACE, in_signature='ss', out_signature='v') @dbus_handle_exceptions def Get(self, interface_name, property_name, sender=None): # pylint: disable=W0613 # get a property interface_name = dbus_to_python(interface_name, str) property_name = dbus_to_python(property_name, str) log.debug1("Get('%s', '%s')", interface_name, property_name) if interface_name == config.dbus.DBUS_INTERFACE: return self._get_property(property_name) elif interface_name in [ config.dbus.DBUS_INTERFACE_ZONE, config.dbus.DBUS_INTERFACE_DIRECT, config.dbus.DBUS_INTERFACE_POLICIES, config.dbus.DBUS_INTERFACE_IPSET ]: raise dbus.exceptions.DBusException( "org.freedesktop.DBus.Error.InvalidArgs: " "Property '%s' does not exist" % property_name) else: raise dbus.exceptions.DBusException( "org.freedesktop.DBus.Error.UnknownInterface: " "Interface '%s' does not exist" % interface_name) @dbus_service_method(dbus.PROPERTIES_IFACE, in_signature='s', out_signature='a{sv}') @dbus_handle_exceptions def GetAll(self, interface_name, sender=None): # pylint: disable=W0613 interface_name = dbus_to_python(interface_name, str) log.debug1("GetAll('%s')", interface_name) ret = { } if interface_name == config.dbus.DBUS_INTERFACE: for x in [ "version", "interface_version", "state", "IPv4", "IPv6", "IPv6_rpfilter", "BRIDGE", "IPSet", "IPSetTypes", "nf_conntrack_helper_setting", "nf_conntrack_helpers", "nf_nat_helpers", "IPv4ICMPTypes", "IPv6ICMPTypes" ]: ret[x] = self._get_property(x) elif interface_name in [ config.dbus.DBUS_INTERFACE_ZONE, config.dbus.DBUS_INTERFACE_DIRECT, config.dbus.DBUS_INTERFACE_POLICIES, config.dbus.DBUS_INTERFACE_IPSET ]: pass else: raise dbus.exceptions.DBusException( "org.freedesktop.DBus.Error.UnknownInterface: " "Interface '%s' does not exist" % interface_name) return dbus.Dictionary(ret, signature="sv") @dbus_polkit_require_auth(config.dbus.PK_ACTION_CONFIG) @dbus_service_method(dbus.PROPERTIES_IFACE, in_signature='ssv') @dbus_handle_exceptions def Set(self, interface_name, property_name, new_value, sender=None): interface_name = dbus_to_python(interface_name, str) property_name = dbus_to_python(property_name, str) new_value = dbus_to_python(new_value) log.debug1("Set('%s', '%s', '%s')", interface_name, property_name, new_value) self.accessCheck(sender) if interface_name == config.dbus.DBUS_INTERFACE: if property_name in [ "version", "interface_version", "state", "IPv4", "IPv6", "IPv6_rpfilter", "BRIDGE", "IPSet", "IPSetTypes", "nf_conntrack_helper_setting", "nf_conntrack_helpers", "nf_nat_helpers", "IPv4ICMPTypes", "IPv6ICMPTypes" ]: raise dbus.exceptions.DBusException( "org.freedesktop.DBus.Error.PropertyReadOnly: " "Property '%s' is read-only" % property_name) else: raise dbus.exceptions.DBusException( "org.freedesktop.DBus.Error.InvalidArgs: " "Property '%s' does not exist" % property_name) elif interface_name in [ config.dbus.DBUS_INTERFACE_ZONE, config.dbus.DBUS_INTERFACE_DIRECT, config.dbus.DBUS_INTERFACE_POLICIES, config.dbus.DBUS_INTERFACE_IPSET ]: raise dbus.exceptions.DBusException( "org.freedesktop.DBus.Error.InvalidArgs: " "Property '%s' does not exist" % property_name) else: raise dbus.exceptions.DBusException( "org.freedesktop.DBus.Error.UnknownInterface: " "Interface '%s' does not exist" % interface_name) @dbus.service.signal(dbus.PROPERTIES_IFACE, signature='sa{sv}as') def PropertiesChanged(self, interface_name, changed_properties, invalidated_properties): interface_name = dbus_to_python(interface_name, str) changed_properties = dbus_to_python(changed_properties) invalidated_properties = dbus_to_python(invalidated_properties) log.debug1("PropertiesChanged('%s', '%s', '%s')", interface_name, changed_properties, invalidated_properties) @dbus_polkit_require_auth(config.dbus.PK_ACTION_INFO) @dbus_service_method(dbus.INTROSPECTABLE_IFACE, out_signature='s') @dbus_handle_exceptions def Introspect(self, sender=None): # pylint: disable=W0613 log.debug2("Introspect()") data = super(FirewallD, self).Introspect(self.path, self.busname.get_bus()) data = dbus_introspection_add_properties(self, data, config.dbus.DBUS_INTERFACE) for interface in [config.dbus.DBUS_INTERFACE_DIRECT]: data = dbus_introspection_add_deprecated(self, data, interface, dbus_service_method_deprecated().deprecated, dbus_service_signal_deprecated().deprecated) return data # reload @dbus_polkit_require_auth(config.dbus.PK_ACTION_CONFIG) @dbus_service_method(config.dbus.DBUS_INTERFACE, in_signature='', out_signature='') @dbus_handle_exceptions def reload(self, sender=None): # pylint: disable=W0613 """Reload the firewall rules. """ log.debug1("reload()") self.fw.reload() self.config.reload() self.Reloaded() # complete_reload @dbus_polkit_require_auth(config.dbus.PK_ACTION_CONFIG) @dbus_service_method(config.dbus.DBUS_INTERFACE, in_signature='', out_signature='') @dbus_handle_exceptions def completeReload(self, sender=None): # pylint: disable=W0613 """Completely reload the firewall. Completely reload the firewall: Stops firewall, unloads modules and starts the firewall again. """ log.debug1("completeReload()") self.fw.reload(True) self.config.reload() self.Reloaded() @dbus.service.signal(config.dbus.DBUS_INTERFACE) @dbus_handle_exceptions def Reloaded(self): log.debug1("Reloaded()") @dbus_polkit_require_auth(config.dbus.PK_ACTION_CONFIG) @dbus_service_method(config.dbus.DBUS_INTERFACE, in_signature='', out_signature= '') def resetToDefaults(self, sender=None): """reset to firewall's builtin defaults. Reloads firewalld to apply changes properly """ log.debug1("firewalld.reset_to_defaults()") self.fw.config.reset_defaults() self.reload() @dbus_polkit_require_auth(config.dbus.PK_ACTION_CONFIG) @dbus_service_method(config.dbus.DBUS_INTERFACE, in_signature='', out_signature='') @dbus_handle_exceptions def checkPermanentConfig(self, sender=None): # pylint: disable=W0613 """Check permanent configuration """ log.debug1("checkPermanentConfig()") check_on_disk_config(self.fw) # runtime to permanent @dbus_polkit_require_auth(config.dbus.PK_ACTION_CONFIG) @dbus_service_method(config.dbus.DBUS_INTERFACE, in_signature='', out_signature='') @dbus_handle_exceptions def runtimeToPermanent(self, sender=None): # pylint: disable=W0613 """Make runtime configuration permanent """ log.debug1("copyRuntimeToPermanent()") if self.fw._state == "FAILED": raise FirewallError(errors.RUNNING_BUT_FAILED, "Saving runtime to permanent is not allowed while " "firewalld is in FAILED state. The permanent " "configuration must be fixed and then firewalld " "restarted. Try `firewall-offline-cmd --check-config`.") error = False # Services or icmptypes can not be modified in runtime, but they can # be removed or modified in permanent environment. Therefore copying # of services and icmptypes to permanent is also needed. # services config_names = self.config.getServiceNames() for name in self.fw.service.get_services(): conf = self.getServiceSettings(name) try: if name in config_names: conf_obj = self.config.getServiceByName(name) if conf_obj.getSettings() != conf: log.debug1("Copying service '%s' settings" % name) conf_obj.update(conf) else: log.debug1("Service '%s' is identical, ignoring." % name) else: log.debug1("Creating service '%s'" % name) self.config.addService(name, conf) except Exception as e: log.warning( "Runtime To Permanent failed on service '%s': %s" % \ (name, e)) error = True # icmptypes config_names = self.config.getIcmpTypeNames() for name in self.fw.icmptype.get_icmptypes(): conf = self.getIcmpTypeSettings(name) try: if name in config_names: conf_obj = self.config.getIcmpTypeByName(name) if conf_obj.getSettings() != conf: log.debug1("Copying icmptype '%s' settings" % name) conf_obj.update(conf) else: log.debug1("IcmpType '%s' is identical, ignoring." % name) else: log.debug1("Creating icmptype '%s'" % name) self.config.addIcmpType(name, conf) except Exception as e: log.warning( "Runtime To Permanent failed on icmptype '%s': %s" % \ (name, e)) error = True # ipsets config_names = self.config.getIPSetNames() for name in self.fw.ipset.get_ipsets(): try: conf = self.getIPSetSettings(name) if name in config_names: conf_obj = self.config.getIPSetByName(name) if conf_obj.getSettings() != conf: log.debug1("Copying ipset '%s' settings" % name) conf_obj.update(conf) else: log.debug1("IPSet '%s' is identical, ignoring." % name) else: log.debug1("Creating ipset '%s'" % name) self.config.addIPSet(name, conf) except Exception as e: log.warning( "Runtime To Permanent failed on ipset '%s': %s" % \ (name, e)) error = True # zones config_names = self.config.getZoneNames() for name in self.fw.zone.get_zones(): conf = self.getZoneSettings2(name) settings = FirewallClientZoneSettings(copy.deepcopy(conf)) changed = False for interface in settings.getInterfaces(): if interface in self.fw._nm_assigned_interfaces: log.debug1("Zone '%s': interface binding for '%s' has been added by NM, ignoring." % (name, interface)) settings.removeInterface(interface) changed = True # For the remaining interfaces, attempt to let NM manage them for interface in settings.getInterfaces(): try: connection = nm_get_connection_of_interface(interface) if connection and nm_set_zone_of_connection(name, connection): settings.removeInterface(interface) changed = True except Exception: pass if changed: conf = settings.getSettingsDict() # For the remaining try to update the ifcfg files for interface in settings.getInterfaces(): ifcfg_set_zone_of_interface(name, interface) try: if name in config_names: conf_obj = self.config.getZoneByName(name) log.debug1("Copying zone '%s' settings" % name) conf_obj.update2(conf) else: log.debug1("Creating zone '%s'" % name) self.config.addZone2(name, conf) except Exception as e: log.warning( "Runtime To Permanent failed on zone '%s': %s" % \ (name, e)) error = True # policies config_names = self.config.getPolicyNames() for name in self.fw.policy.get_policies_not_derived_from_zone(): conf = self.getPolicySettings(name) try: if name in config_names: conf_obj = self.config.getPolicyByName(name) conf_obj.update(conf) else: log.debug1("Creating policy '%s'" % name) self.config.addPolicy(name, conf) except Exception as e: log.warning( "Runtime To Permanent failed on policy '%s': %s" % \ (name, e)) error = True # helpers config_names = self.config.getHelperNames() for name in self.fw.helper.get_helpers(): conf = self.getHelperSettings(name) try: if name in config_names: conf_obj = self.config.getHelperByName(name) if conf_obj.getSettings() != conf: log.debug1("Copying helper '%s' settings" % name) conf_obj.update(conf) else: log.debug1("Helper '%s' is identical, ignoring." % name) else: log.debug1("Creating helper '%s'" % name) self.config.addHelper(name, conf) except Exception as e: log.warning( "Runtime To Permanent failed on helper '%s': %s" % \ (name, e)) error = True # direct # rt_config = self.fw.direct.get_config() conf = ( self.fw.direct.get_all_chains(), self.fw.direct.get_all_rules(), self.fw.direct.get_all_passthroughs() ) try: if self.config.getSettings() != conf: log.debug1("Copying direct configuration") self.config.update(conf) else: log.debug1("Direct configuration is identical, ignoring.") except Exception as e: log.warning( "Runtime To Permanent failed on direct configuration: %s" % e) error = True # policies conf = self.fw.policies.lockdown_whitelist.export_config() try: if self.config.getSettings() != conf: log.debug1("Copying policies configuration") self.config.setLockdownWhitelist(conf) else: log.debug1("Policies configuration is identical, ignoring.") except Exception as e: log.warning( "Runtime To Permanent failed on policies configuration: %s" % \ e) error = True if error: raise FirewallError(errors.RT_TO_PERM_FAILED) # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # POLICIES # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # lockdown @dbus_polkit_require_auth(config.dbus.PK_ACTION_POLICIES) @dbus_service_method(config.dbus.DBUS_INTERFACE_POLICIES, in_signature='', out_signature='') @dbus_handle_exceptions def enableLockdown(self, sender=None): """Enable lockdown policies """ log.debug1("policies.enableLockdown()") self.accessCheck(sender) self.fw.policies.enable_lockdown() self.LockdownEnabled() @dbus_polkit_require_auth(config.dbus.PK_ACTION_POLICIES) @dbus_service_method(config.dbus.DBUS_INTERFACE_POLICIES, in_signature='', out_signature='') @dbus_handle_exceptions def disableLockdown(self, sender=None): """Disable lockdown policies """ log.debug1("policies.disableLockdown()") self.accessCheck(sender) self.fw.policies.disable_lockdown() self.LockdownDisabled() @dbus_polkit_require_auth(config.dbus.PK_ACTION_POLICIES_INFO) @dbus_service_method(config.dbus.DBUS_INTERFACE_POLICIES, in_signature='', out_signature='b') @dbus_handle_exceptions def queryLockdown(self, sender=None): # pylint: disable=W0613 """Returns True if lockdown is enabled """ log.debug1("policies.queryLockdown()") # no access check here return self.fw.policies.query_lockdown() @dbus.service.signal(config.dbus.DBUS_INTERFACE_POLICIES, signature='') @dbus_handle_exceptions def LockdownEnabled(self): log.debug1("LockdownEnabled()") @dbus.service.signal(config.dbus.DBUS_INTERFACE_POLICIES, signature='') @dbus_handle_exceptions def LockdownDisabled(self): log.debug1("LockdownDisabled()") # lockdown whitelist # command @dbus_polkit_require_auth(config.dbus.PK_ACTION_POLICIES) @dbus_service_method(config.dbus.DBUS_INTERFACE_POLICIES, in_signature='s', out_signature='') @dbus_handle_exceptions def addLockdownWhitelistCommand(self, command, sender=None): """Add lockdown command """ command = dbus_to_python(command, str) log.debug1("policies.addLockdownWhitelistCommand('%s')" % command) self.accessCheck(sender) self.fw.policies.lockdown_whitelist.add_command(command) self.LockdownWhitelistCommandAdded(command) @dbus_polkit_require_auth(config.dbus.PK_ACTION_POLICIES) @dbus_service_method(config.dbus.DBUS_INTERFACE_POLICIES, in_signature='s', out_signature='') @dbus_handle_exceptions def removeLockdownWhitelistCommand(self, command, sender=None): """Remove lockdown command """ command = dbus_to_python(command, str) log.debug1("policies.removeLockdownWhitelistCommand('%s')" % command) self.accessCheck(sender) self.fw.policies.lockdown_whitelist.remove_command(command) self.LockdownWhitelistCommandRemoved(command) @dbus_polkit_require_auth(config.dbus.PK_ACTION_POLICIES_INFO) @dbus_service_method(config.dbus.DBUS_INTERFACE_POLICIES, in_signature='s', out_signature='b') @dbus_handle_exceptions def queryLockdownWhitelistCommand(self, command, sender=None): # pylint: disable=W0613 """Query lockdown command """ command = dbus_to_python(command, str) log.debug1("policies.queryLockdownWhitelistCommand('%s')" % command) # no access check here return self.fw.policies.lockdown_whitelist.has_command(command) @dbus_polkit_require_auth(config.dbus.PK_ACTION_POLICIES_INFO) @dbus_service_method(config.dbus.DBUS_INTERFACE_POLICIES, in_signature='', out_signature='as') @dbus_handle_exceptions def getLockdownWhitelistCommands(self, sender=None): # pylint: disable=W0613 """Add lockdown command """ log.debug1("policies.getLockdownWhitelistCommands()") # no access check here return self.fw.policies.lockdown_whitelist.get_commands() @dbus.service.signal(config.dbus.DBUS_INTERFACE_POLICIES, signature='s') @dbus_handle_exceptions def LockdownWhitelistCommandAdded(self, command): log.debug1("LockdownWhitelistCommandAdded('%s')" % command) @dbus.service.signal(config.dbus.DBUS_INTERFACE_POLICIES, signature='s') @dbus_handle_exceptions def LockdownWhitelistCommandRemoved(self, command): log.debug1("LockdownWhitelistCommandRemoved('%s')" % command) # uid @dbus_polkit_require_auth(config.dbus.PK_ACTION_POLICIES) @dbus_service_method(config.dbus.DBUS_INTERFACE_POLICIES, in_signature='i', out_signature='') @dbus_handle_exceptions def addLockdownWhitelistUid(self, uid, sender=None): """Add lockdown uid """ uid = dbus_to_python(uid, int) log.debug1("policies.addLockdownWhitelistUid('%s')" % uid) self.accessCheck(sender) self.fw.policies.lockdown_whitelist.add_uid(uid) self.LockdownWhitelistUidAdded(uid) @dbus_polkit_require_auth(config.dbus.PK_ACTION_POLICIES) @dbus_service_method(config.dbus.DBUS_INTERFACE_POLICIES, in_signature='i', out_signature='') @dbus_handle_exceptions def removeLockdownWhitelistUid(self, uid, sender=None): """Remove lockdown uid """ uid = dbus_to_python(uid, int) log.debug1("policies.removeLockdownWhitelistUid('%s')" % uid) self.accessCheck(sender) self.fw.policies.lockdown_whitelist.remove_uid(uid) self.LockdownWhitelistUidRemoved(uid) @dbus_polkit_require_auth(config.dbus.PK_ACTION_POLICIES_INFO) @dbus_service_method(config.dbus.DBUS_INTERFACE_POLICIES, in_signature='i', out_signature='b') @dbus_handle_exceptions def queryLockdownWhitelistUid(self, uid, sender=None): # pylint: disable=W0613 """Query lockdown uid """ uid = dbus_to_python(uid, int) log.debug1("policies.queryLockdownWhitelistUid('%s')" % uid) # no access check here return self.fw.policies.lockdown_whitelist.has_uid(uid) @dbus_polkit_require_auth(config.dbus.PK_ACTION_POLICIES_INFO) @dbus_service_method(config.dbus.DBUS_INTERFACE_POLICIES, in_signature='', out_signature='ai') @dbus_handle_exceptions def getLockdownWhitelistUids(self, sender=None): # pylint: disable=W0613 """Add lockdown uid """ log.debug1("policies.getLockdownWhitelistUids()") # no access check here return self.fw.policies.lockdown_whitelist.get_uids() @dbus.service.signal(config.dbus.DBUS_INTERFACE_POLICIES, signature='i') @dbus_handle_exceptions def LockdownWhitelistUidAdded(self, uid): log.debug1("LockdownWhitelistUidAdded(%d)" % uid) @dbus.service.signal(config.dbus.DBUS_INTERFACE_POLICIES, signature='i') @dbus_handle_exceptions def LockdownWhitelistUidRemoved(self, uid): log.debug1("LockdownWhitelistUidRemoved(%d)" % uid) # user @dbus_polkit_require_auth(config.dbus.PK_ACTION_POLICIES) @dbus_service_method(config.dbus.DBUS_INTERFACE_POLICIES, in_signature='s', out_signature='') @dbus_handle_exceptions def addLockdownWhitelistUser(self, user, sender=None): """Add lockdown user """ user = dbus_to_python(user, str) log.debug1("policies.addLockdownWhitelistUser('%s')" % user) self.accessCheck(sender) self.fw.policies.lockdown_whitelist.add_user(user) self.LockdownWhitelistUserAdded(user) @dbus_polkit_require_auth(config.dbus.PK_ACTION_POLICIES) @dbus_service_method(config.dbus.DBUS_INTERFACE_POLICIES, in_signature='s', out_signature='') @dbus_handle_exceptions def removeLockdownWhitelistUser(self, user, sender=None): """Remove lockdown user """ user = dbus_to_python(user, str) log.debug1("policies.removeLockdownWhitelistUser('%s')" % user) self.accessCheck(sender) self.fw.policies.lockdown_whitelist.remove_user(user) self.LockdownWhitelistUserRemoved(user) @dbus_polkit_require_auth(config.dbus.PK_ACTION_POLICIES_INFO) @dbus_service_method(config.dbus.DBUS_INTERFACE_POLICIES, in_signature='s', out_signature='b') @dbus_handle_exceptions def queryLockdownWhitelistUser(self, user, sender=None): # pylint: disable=W0613 """Query lockdown user """ user = dbus_to_python(user, str) log.debug1("policies.queryLockdownWhitelistUser('%s')" % user) # no access check here return self.fw.policies.lockdown_whitelist.has_user(user) @dbus_polkit_require_auth(config.dbus.PK_ACTION_POLICIES_INFO) @dbus_service_method(config.dbus.DBUS_INTERFACE_POLICIES, in_signature='', out_signature='as') @dbus_handle_exceptions def getLockdownWhitelistUsers(self, sender=None): # pylint: disable=W0613 """Add lockdown user """ log.debug1("policies.getLockdownWhitelistUsers()") # no access check here return self.fw.policies.lockdown_whitelist.get_users() @dbus.service.signal(config.dbus.DBUS_INTERFACE_POLICIES, signature='s') @dbus_handle_exceptions def LockdownWhitelistUserAdded(self, user): log.debug1("LockdownWhitelistUserAdded('%s')" % user) @dbus.service.signal(config.dbus.DBUS_INTERFACE_POLICIES, signature='s') @dbus_handle_exceptions def LockdownWhitelistUserRemoved(self, user): log.debug1("LockdownWhitelistUserRemoved('%s')" % user) # context @dbus_polkit_require_auth(config.dbus.PK_ACTION_POLICIES) @dbus_service_method(config.dbus.DBUS_INTERFACE_POLICIES, in_signature='s', out_signature='') @dbus_handle_exceptions def addLockdownWhitelistContext(self, context, sender=None): """Add lockdown context """ context = dbus_to_python(context, str) log.debug1("policies.addLockdownWhitelistContext('%s')" % context) self.accessCheck(sender) self.fw.policies.lockdown_whitelist.add_context(context) self.LockdownWhitelistContextAdded(context) @dbus_polkit_require_auth(config.dbus.PK_ACTION_POLICIES) @dbus_service_method(config.dbus.DBUS_INTERFACE_POLICIES, in_signature='s', out_signature='') @dbus_handle_exceptions def removeLockdownWhitelistContext(self, context, sender=None): """Remove lockdown context """ context = dbus_to_python(context, str) log.debug1("policies.removeLockdownWhitelistContext('%s')" % context) self.accessCheck(sender) self.fw.policies.lockdown_whitelist.remove_context(context) self.LockdownWhitelistContextRemoved(context) @dbus_polkit_require_auth(config.dbus.PK_ACTION_POLICIES_INFO) @dbus_service_method(config.dbus.DBUS_INTERFACE_POLICIES, in_signature='s', out_signature='b') @dbus_handle_exceptions def queryLockdownWhitelistContext(self, context, sender=None): # pylint: disable=W0613 """Query lockdown context """ context = dbus_to_python(context, str) log.debug1("policies.queryLockdownWhitelistContext('%s')" % context) # no access check here return self.fw.policies.lockdown_whitelist.has_context(context) @dbus_polkit_require_auth(config.dbus.PK_ACTION_POLICIES_INFO) @dbus_service_method(config.dbus.DBUS_INTERFACE_POLICIES, in_signature='', out_signature='as') @dbus_handle_exceptions def getLockdownWhitelistContexts(self, sender=None): # pylint: disable=W0613 """Add lockdown context """ log.debug1("policies.getLockdownWhitelistContexts()") # no access check here return self.fw.policies.lockdown_whitelist.get_contexts() @dbus.service.signal(config.dbus.DBUS_INTERFACE_POLICIES, signature='s') @dbus_handle_exceptions def LockdownWhitelistContextAdded(self, context): log.debug1("LockdownWhitelistContextAdded('%s')" % context) @dbus.service.signal(config.dbus.DBUS_INTERFACE_POLICIES, signature='s') @dbus_handle_exceptions def LockdownWhitelistContextRemoved(self, context): log.debug1("LockdownWhitelistContextRemoved('%s')" % context) # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # PANIC @dbus_polkit_require_auth(config.dbus.PK_ACTION_CONFIG) @dbus_service_method(config.dbus.DBUS_INTERFACE, in_signature='', out_signature='') @dbus_handle_exceptions def enablePanicMode(self, sender=None): """Enable panic mode. All ingoing and outgoing connections and packets will be blocked. """ log.debug1("enablePanicMode()") self.accessCheck(sender) self.fw.enable_panic_mode() self.PanicModeEnabled() @dbus_polkit_require_auth(config.dbus.PK_ACTION_CONFIG) @dbus_service_method(config.dbus.DBUS_INTERFACE, in_signature='', out_signature='') @dbus_handle_exceptions def disablePanicMode(self, sender=None): """Disable panic mode. Enables normal mode: Allowed ingoing and outgoing connections will not be blocked anymore """ log.debug1("disablePanicMode()") self.accessCheck(sender) self.fw.disable_panic_mode() self.PanicModeDisabled() @dbus_polkit_require_auth(config.dbus.PK_ACTION_INFO) @dbus_service_method(config.dbus.DBUS_INTERFACE, in_signature='', out_signature='b') @dbus_handle_exceptions def queryPanicMode(self, sender=None): # pylint: disable=W0613 # returns True if in panic mode log.debug1("queryPanicMode()") return self.fw.query_panic_mode() @dbus.service.signal(config.dbus.DBUS_INTERFACE, signature='') @dbus_handle_exceptions def PanicModeEnabled(self): log.debug1("PanicModeEnabled()") @dbus.service.signal(config.dbus.DBUS_INTERFACE, signature='') @dbus_handle_exceptions def PanicModeDisabled(self): log.debug1("PanicModeDisabled()") # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # list functions @dbus_polkit_require_auth(config.dbus.PK_ACTION_CONFIG_INFO) @dbus_service_method(config.dbus.DBUS_INTERFACE, in_signature='s', out_signature="(sssbsasa(ss)asba(ssss)asasasasa(ss)b)") @dbus_handle_exceptions def getZoneSettings(self, zone, sender=None): # pylint: disable=W0613 # returns zone settings for zone zone = dbus_to_python(zone, str) log.debug1("getZoneSettings(%s)", zone) return self.fw.zone.get_config_with_settings(zone) @dbus_polkit_require_auth(config.dbus.PK_ACTION_CONFIG_INFO) @dbus_service_method(config.dbus.DBUS_INTERFACE_ZONE, in_signature='s', out_signature="a{sv}") @dbus_handle_exceptions def getZoneSettings2(self, zone, sender=None): zone = dbus_to_python(zone, str) log.debug1("getZoneSettings2(%s)", zone) return self.fw.zone.get_config_with_settings_dict(zone) @dbus_polkit_require_auth(config.dbus.PK_ACTION_CONFIG_INFO) @dbus_service_method(config.dbus.DBUS_INTERFACE_ZONE, in_signature='sa{sv}') @dbus_handle_exceptions def setZoneSettings2(self, zone, settings, sender=None): zone = dbus_to_python(zone, str) log.debug1("setZoneSettings2(%s)", zone) self.accessCheck(sender) self.fw.zone.set_config_with_settings_dict(zone, dbus_to_python(settings), sender) self.ZoneUpdated(zone, settings) @dbus.service.signal(config.dbus.DBUS_INTERFACE_ZONE, signature='sa{sv}') @dbus_handle_exceptions def ZoneUpdated(self, zone, settings): log.debug1("zone.ZoneUpdated('%s', '%s')" % (zone, settings)) @dbus_polkit_require_auth(config.dbus.PK_ACTION_CONFIG_INFO) @dbus_service_method(config.dbus.DBUS_INTERFACE_POLICY, in_signature='s', out_signature="a{sv}") @dbus_handle_exceptions def getPolicySettings(self, policy, sender=None): policy = dbus_to_python(policy, str) log.debug1("policy.getPolicySettings(%s)", policy) return self.fw.policy.get_config_with_settings_dict(policy) @dbus_polkit_require_auth(config.dbus.PK_ACTION_CONFIG_INFO) @dbus_service_method(config.dbus.DBUS_INTERFACE_POLICY, in_signature='sa{sv}') @dbus_handle_exceptions def setPolicySettings(self, policy, settings, sender=None): policy = dbus_to_python(policy, str) log.debug1("policy.setPolicySettings(%s)", policy) self.accessCheck(sender) self.fw.policy.set_config_with_settings_dict(policy, dbus_to_python(settings), sender) self.PolicyUpdated(policy, settings) @dbus.service.signal(config.dbus.DBUS_INTERFACE_POLICY, signature='sa{sv}') @dbus_handle_exceptions def PolicyUpdated(self, policy, settings): log.debug1("policy.PolicyUpdated('%s', '%s')" % (policy, settings)) @dbus_polkit_require_auth(config.dbus.PK_ACTION_INFO) @dbus_service_method(config.dbus.DBUS_INTERFACE, in_signature='', out_signature='as') @dbus_handle_exceptions def listServices(self, sender=None): # pylint: disable=W0613 # returns the list of services # TODO: should be renamed to getServices() # because is called by firewall-cmd --get-services log.debug1("listServices()") return self.fw.service.get_services() @dbus_polkit_require_auth(config.dbus.PK_ACTION_CONFIG_INFO) @dbus_service_method(config.dbus.DBUS_INTERFACE, in_signature='s', out_signature='(sssa(ss)asa{ss}asa(ss))') @dbus_handle_exceptions def getServiceSettings(self, service, sender=None): # pylint: disable=W0613 # returns service settings for service service = dbus_to_python(service, str) log.debug1("getServiceSettings(%s)", service) obj = self.fw.service.get_service(service) conf_dict = obj.export_config_dict() conf_list = [] for i in range(8): # tuple based dbus API has 8 elements if obj.IMPORT_EXPORT_STRUCTURE[i][0] not in conf_dict: # old API needs the empty elements as well. Grab it from the # object otherwise we don't know the type. conf_list.append(copy.deepcopy(getattr(obj, obj.IMPORT_EXPORT_STRUCTURE[i][0]))) else: conf_list.append(conf_dict[obj.IMPORT_EXPORT_STRUCTURE[i][0]]) return tuple(conf_list) @dbus_polkit_require_auth(config.dbus.PK_ACTION_CONFIG_INFO) @dbus_service_method(config.dbus.DBUS_INTERFACE, in_signature='s', out_signature='a{sv}') @dbus_handle_exceptions def getServiceSettings2(self, service, sender=None): # pylint: disable=W0613 service = dbus_to_python(service, str) log.debug1("getServiceSettings2(%s)", service) obj = self.fw.service.get_service(service) return obj.export_config_dict() @dbus_polkit_require_auth(config.dbus.PK_ACTION_INFO) @dbus_service_method(config.dbus.DBUS_INTERFACE, in_signature='', out_signature='as') @dbus_handle_exceptions def listIcmpTypes(self, sender=None): # pylint: disable=W0613 # returns the list of services # TODO: should be renamed to getIcmptypes() # because is called by firewall-cmd --get-icmptypes log.debug1("listIcmpTypes()") return self.fw.icmptype.get_icmptypes() @dbus_polkit_require_auth(config.dbus.PK_ACTION_CONFIG_INFO) @dbus_service_method(config.dbus.DBUS_INTERFACE, in_signature='s', out_signature=IcmpType.DBUS_SIGNATURE) @dbus_handle_exceptions def getIcmpTypeSettings(self, icmptype, sender=None): # pylint: disable=W0613 # returns icmptype settings for icmptype icmptype = dbus_to_python(icmptype, str) log.debug1("getIcmpTypeSettings(%s)", icmptype) return self.fw.icmptype.get_icmptype(icmptype).export_config() # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # LOG DENIED @dbus_polkit_require_auth(config.dbus.PK_ACTION_CONFIG_INFO) @dbus_service_method(config.dbus.DBUS_INTERFACE, in_signature='', out_signature='s') @dbus_handle_exceptions def getLogDenied(self, sender=None): # pylint: disable=W0613 # returns the log denied value log.debug1("getLogDenied()") return self.fw.get_log_denied() @dbus_polkit_require_auth(config.dbus.PK_ACTION_CONFIG) @dbus_service_method(config.dbus.DBUS_INTERFACE, in_signature='s', out_signature='') @dbus_handle_exceptions def setLogDenied(self, value, sender=None): # set the log denied value value = dbus_to_python(value, str) log.debug1("setLogDenied('%s')" % value) self.accessCheck(sender) self.fw.set_log_denied(value) self.LogDeniedChanged(value) # must reload the firewall as well self.fw.reload() self.config.reload() self.Reloaded() @dbus.service.signal(config.dbus.DBUS_INTERFACE, signature='s') @dbus_handle_exceptions def LogDeniedChanged(self, value): log.debug1("LogDeniedChanged('%s')" % (value)) # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # AUTOMATIC HELPER ASSIGNMENT @dbus_polkit_require_auth(config.dbus.PK_ACTION_CONFIG_INFO) @dbus_service_method(config.dbus.DBUS_INTERFACE, in_signature='', out_signature='s') @dbus_handle_exceptions def getAutomaticHelpers(self, sender=None): # pylint: disable=W0613 # returns the automatic helpers value log.debug1("getAutomaticHelpers()") # NOTE: This feature was removed and is now a noop. We retain the dbus # call to keep API. return "no" @dbus_polkit_require_auth(config.dbus.PK_ACTION_CONFIG) @dbus_service_method(config.dbus.DBUS_INTERFACE, in_signature='s', out_signature='') @dbus_handle_exceptions def setAutomaticHelpers(self, value, sender=None): # set the automatic helpers value value = dbus_to_python(value, str) log.debug1("setAutomaticHelpers('%s')" % value) self.accessCheck(sender) # NOTE: This feature was removed and is now a noop. We retain the dbus # call to keep API. @dbus.service.signal(config.dbus.DBUS_INTERFACE, signature='s') @dbus_handle_exceptions def AutomaticHelpersChanged(self, value): log.debug1("AutomaticHelpersChanged('%s')" % (value)) # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # DEFAULT ZONE @dbus_polkit_require_auth(config.dbus.PK_ACTION_INFO) @dbus_service_method(config.dbus.DBUS_INTERFACE, in_signature='', out_signature='s') @dbus_handle_exceptions def getDefaultZone(self, sender=None): # pylint: disable=W0613 # returns the system default zone log.debug1("getDefaultZone()") return self.fw.get_default_zone() @dbus_polkit_require_auth(config.dbus.PK_ACTION_CONFIG) @dbus_service_method(config.dbus.DBUS_INTERFACE, in_signature='s', out_signature='') @dbus_handle_exceptions def setDefaultZone(self, zone, sender=None): # set the system default zone zone = dbus_to_python(zone, str) log.debug1("setDefaultZone('%s')" % zone) self.accessCheck(sender) self.fw.set_default_zone(zone) self.DefaultZoneChanged(zone) @dbus.service.signal(config.dbus.DBUS_INTERFACE, signature='s') @dbus_handle_exceptions def DefaultZoneChanged(self, zone): log.debug1("DefaultZoneChanged('%s')" % (zone)) # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # POLICY INTERFACE # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # POLICIES @dbus_polkit_require_auth(config.dbus.PK_ACTION_INFO) @dbus_service_method(config.dbus.DBUS_INTERFACE_POLICY, in_signature='', out_signature='as') @dbus_handle_exceptions def getPolicies(self, sender=None): log.debug1("policy.getPolicies()") return self.fw.policy.get_policies_not_derived_from_zone() @dbus_polkit_require_auth(config.dbus.PK_ACTION_INFO) @dbus_service_method(config.dbus.DBUS_INTERFACE_POLICY, in_signature='', out_signature='a{sa{sas}}') @dbus_handle_exceptions def getActivePolicies(self, sender=None): log.debug1("policy.getActivePolicies()") policies = { } for policy in self.fw.policy.get_active_policies_not_derived_from_zone(): policies[policy] = { } policies[policy]["ingress_zones"] = self.fw.policy.list_ingress_zones(policy) policies[policy]["egress_zones"] = self.fw.policy.list_egress_zones(policy) return policies # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # ZONE INTERFACE # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # ZONES @dbus_polkit_require_auth(config.dbus.PK_ACTION_INFO) # TODO: shouldn't this be in DBUS_INTERFACE instead of DBUS_INTERFACE_ZONE ? @dbus_service_method(config.dbus.DBUS_INTERFACE_ZONE, in_signature='', out_signature='as') @dbus_handle_exceptions def getZones(self, sender=None): # pylint: disable=W0613 # returns the list of zones log.debug1("zone.getZones()") return self.fw.zone.get_zones() @dbus_polkit_require_auth(config.dbus.PK_ACTION_INFO) @dbus_service_method(config.dbus.DBUS_INTERFACE_ZONE, in_signature='', out_signature='a{sa{sas}}') @dbus_handle_exceptions def getActiveZones(self, sender=None): # pylint: disable=W0613 # returns the list of active zones log.debug1("zone.getActiveZones()") zones = { } for zone in self.fw.zone.get_zones(): interfaces = self.fw.zone.list_interfaces(zone) sources = self.fw.zone.list_sources(zone) if len(interfaces) + len(sources) > 0: zones[zone] = { } if len(interfaces) > 0: zones[zone]["interfaces"] = interfaces if len(sources) > 0: zones[zone]["sources"] = sources return zones @dbus_polkit_require_auth(config.dbus.PK_ACTION_INFO) @dbus_service_method(config.dbus.DBUS_INTERFACE_ZONE, in_signature='s', out_signature='s') @dbus_handle_exceptions def getZoneOfInterface(self, interface, sender=None): # pylint: disable=W0613 """Return the zone an interface belongs to. :Parameters: `interface` : str Name of the interface :Returns: str. The name of the zone. """ interface = dbus_to_python(interface, str) log.debug1("zone.getZoneOfInterface('%s')" % interface) zone = self.fw.zone.get_zone_of_interface(interface) if zone: return zone return "" @dbus_polkit_require_auth(config.dbus.PK_ACTION_INFO) @dbus_service_method(config.dbus.DBUS_INTERFACE_ZONE, in_signature='s', out_signature='s') @dbus_handle_exceptions def getZoneOfSource(self, source, sender=None): # pylint: disable=W0613 #Return the zone an source belongs to. source = dbus_to_python(source, str) log.debug1("zone.getZoneOfSource('%s')" % source) zone = self.fw.zone.get_zone_of_source(source) if zone: return zone return "" @dbus_polkit_require_auth(config.dbus.PK_ACTION_CONFIG_INFO) @dbus_service_method(config.dbus.DBUS_INTERFACE_ZONE, in_signature='s', out_signature='b') @dbus_handle_exceptions def isImmutable(self, zone, sender=None): # pylint: disable=W0613 # no immutable zones anymore return False # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # INTERFACES @dbus_polkit_require_auth(config.dbus.PK_ACTION_CONFIG) @dbus_service_method(config.dbus.DBUS_INTERFACE_ZONE, in_signature='ss', out_signature='s') @dbus_handle_exceptions def addInterface(self, zone, interface, sender=None): """Add an interface to a zone. If zone is empty, use default zone. """ zone = dbus_to_python(zone, str) interface = dbus_to_python(interface, str) log.debug1("zone.addInterface('%s', '%s')" % (zone, interface)) self.accessCheck(sender) _zone = self.fw.zone.add_interface(zone, interface, sender) self.InterfaceAdded(_zone, interface) return _zone @dbus_polkit_require_auth(config.dbus.PK_ACTION_CONFIG) @dbus_service_method(config.dbus.DBUS_INTERFACE_ZONE, in_signature='ss', out_signature='s') @dbus_handle_exceptions def changeZone(self, zone, interface, sender=None): """Change a zone an interface is part of. If zone is empty, use default zone. This function is deprecated, use changeZoneOfInterface instead """ zone = dbus_to_python(zone, str) interface = dbus_to_python(interface, str) return self.changeZoneOfInterface(zone, interface, sender) @dbus_polkit_require_auth(config.dbus.PK_ACTION_CONFIG) @dbus_service_method(config.dbus.DBUS_INTERFACE_ZONE, in_signature='ss', out_signature='s') @dbus_handle_exceptions def changeZoneOfInterface(self, zone, interface, sender=None): """Change a zone an interface is part of. If zone is empty, use default zone. """ zone = dbus_to_python(zone, str) interface = dbus_to_python(interface, str) log.debug1("zone.changeZoneOfInterface('%s', '%s')" % (zone, interface)) self.accessCheck(sender) _zone = self.fw.zone.change_zone_of_interface(zone, interface, sender) self.ZoneOfInterfaceChanged(_zone, interface) return _zone @dbus_polkit_require_auth(config.dbus.PK_ACTION_CONFIG) @dbus_service_method(config.dbus.DBUS_INTERFACE_ZONE, in_signature='ss', out_signature='s') @dbus_handle_exceptions def removeInterface(self, zone, interface, sender=None): """Remove interface from a zone. If zone is empty, remove from zone the interface belongs to. """ zone = dbus_to_python(zone, str) interface = dbus_to_python(interface, str) log.debug1("zone.removeInterface('%s', '%s')" % (zone, interface)) self.accessCheck(sender) _zone = self.fw.zone.remove_interface(zone, interface) self.InterfaceRemoved(_zone, interface) return _zone @dbus_polkit_require_auth(config.dbus.PK_ACTION_CONFIG_INFO) @dbus_service_method(config.dbus.DBUS_INTERFACE_ZONE, in_signature='ss', out_signature='b') @dbus_handle_exceptions def queryInterface(self, zone, interface, sender=None): # pylint: disable=W0613 """Return true if an interface is in a zone. If zone is empty, use default zone. """ zone = dbus_to_python(zone, str) interface = dbus_to_python(interface, str) log.debug1("zone.queryInterface('%s', '%s')" % (zone, interface)) return self.fw.zone.query_interface(zone, interface) @dbus_polkit_require_auth(config.dbus.PK_ACTION_CONFIG_INFO) @dbus_service_method(config.dbus.DBUS_INTERFACE_ZONE, in_signature='s', out_signature='as') @dbus_handle_exceptions def getInterfaces(self, zone, sender=None): # pylint: disable=W0613 """Return the list of interfaces of a zone. If zone is empty, use default zone. """ # TODO: should be renamed to listInterfaces() # because is called by firewall-cmd --zone --list-interfaces zone = dbus_to_python(zone, str) log.debug1("zone.getInterfaces('%s')" % (zone)) return self.fw.zone.list_interfaces(zone) @dbus.service.signal(config.dbus.DBUS_INTERFACE_ZONE, signature='ss') @dbus_handle_exceptions def InterfaceAdded(self, zone, interface): log.debug1("zone.InterfaceAdded('%s', '%s')" % (zone, interface)) @dbus.service.signal(config.dbus.DBUS_INTERFACE_ZONE, signature='ss') @dbus_handle_exceptions def ZoneChanged(self, zone, interface): """ This signal is deprecated. """ log.debug1("zone.ZoneChanged('%s', '%s')" % (zone, interface)) @dbus.service.signal(config.dbus.DBUS_INTERFACE_ZONE, signature='ss') @dbus_handle_exceptions def ZoneOfInterfaceChanged(self, zone, interface): log.debug1("zone.ZoneOfInterfaceChanged('%s', '%s')" % (zone, interface)) self.ZoneChanged(zone, interface) @dbus.service.signal(config.dbus.DBUS_INTERFACE_ZONE, signature='ss') @dbus_handle_exceptions def InterfaceRemoved(self, zone, interface): log.debug1("zone.InterfaceRemoved('%s', '%s')" % (zone, interface)) # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # SOURCES @dbus_polkit_require_auth(config.dbus.PK_ACTION_CONFIG) @dbus_service_method(config.dbus.DBUS_INTERFACE_ZONE, in_signature='ss', out_signature='s') @dbus_handle_exceptions def addSource(self, zone, source, sender=None): """Add a source to a zone. If zone is empty, use default zone. """ zone = dbus_to_python(zone, str) source = dbus_to_python(source, str) log.debug1("zone.addSource('%s', '%s')" % (zone, source)) self.accessCheck(sender) _zone = self.fw.zone.add_source(zone, source, sender) self.SourceAdded(_zone, source) return _zone @dbus_polkit_require_auth(config.dbus.PK_ACTION_CONFIG) @dbus_service_method(config.dbus.DBUS_INTERFACE_ZONE, in_signature='ss', out_signature='s') @dbus_handle_exceptions def changeZoneOfSource(self, zone, source, sender=None): """Change a zone an source is part of. If zone is empty, use default zone. """ zone = dbus_to_python(zone, str) source = dbus_to_python(source, str) log.debug1("zone.changeZoneOfSource('%s', '%s')" % (zone, source)) self.accessCheck(sender) _zone = self.fw.zone.change_zone_of_source(zone, source, sender) self.ZoneOfSourceChanged(_zone, source) return _zone @dbus_polkit_require_auth(config.dbus.PK_ACTION_CONFIG) @dbus_service_method(config.dbus.DBUS_INTERFACE_ZONE, in_signature='ss', out_signature='s') @dbus_handle_exceptions def removeSource(self, zone, source, sender=None): """Remove source from a zone. If zone is empty, remove from zone the source belongs to. """ zone = dbus_to_python(zone, str) source = dbus_to_python(source, str) log.debug1("zone.removeSource('%s', '%s')" % (zone, source)) self.accessCheck(sender) _zone = self.fw.zone.remove_source(zone, source) self.SourceRemoved(_zone, source) return _zone @dbus_polkit_require_auth(config.dbus.PK_ACTION_CONFIG_INFO) @dbus_service_method(config.dbus.DBUS_INTERFACE_ZONE, in_signature='ss', out_signature='b') @dbus_handle_exceptions def querySource(self, zone, source, sender=None): # pylint: disable=W0613 """Return true if an source is in a zone. If zone is empty, use default zone. """ zone = dbus_to_python(zone, str) source = dbus_to_python(source, str) log.debug1("zone.querySource('%s', '%s')" % (zone, source)) return self.fw.zone.query_source(zone, source) @dbus_polkit_require_auth(config.dbus.PK_ACTION_CONFIG_INFO) @dbus_service_method(config.dbus.DBUS_INTERFACE_ZONE, in_signature='s', out_signature='as') @dbus_handle_exceptions def getSources(self, zone, sender=None): # pylint: disable=W0613 """Return the list of sources of a zone. If zone is empty, use default zone. """ # TODO: should be renamed to listSources() # because is called by firewall-cmd --zone --list-sources zone = dbus_to_python(zone, str) log.debug1("zone.getSources('%s')" % (zone)) return self.fw.zone.list_sources(zone) @dbus.service.signal(config.dbus.DBUS_INTERFACE_ZONE, signature='ss') @dbus_handle_exceptions def SourceAdded(self, zone, source): log.debug1("zone.SourceAdded('%s', '%s')" % (zone, source)) @dbus.service.signal(config.dbus.DBUS_INTERFACE_ZONE, signature='ss') @dbus_handle_exceptions def ZoneOfSourceChanged(self, zone, source): log.debug1("zone.ZoneOfSourceChanged('%s', '%s')" % (zone, source)) @dbus.service.signal(config.dbus.DBUS_INTERFACE_ZONE, signature='ss') @dbus_handle_exceptions def SourceRemoved(self, zone, source): log.debug1("zone.SourceRemoved('%s', '%s')" % (zone, source)) # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # RICH RULES @dbus_handle_exceptions def disableTimedRichRule(self, zone, rule): log.debug1("zone.disableTimedRichRule('%s', '%s')" % (zone, rule)) del self._timeouts[zone][rule] obj = Rich_Rule(rule_str=rule) self.fw.zone.remove_rule(zone, obj) self.RichRuleRemoved(zone, rule) @dbus_polkit_require_auth(config.dbus.PK_ACTION_CONFIG) @dbus_service_method(config.dbus.DBUS_INTERFACE_ZONE, in_signature='ssi', out_signature='s') @dbus_handle_exceptions def addRichRule(self, zone, rule, timeout, sender=None): # pylint: disable=W0613 zone = dbus_to_python(zone, str) rule = dbus_to_python(rule, str) timeout = dbus_to_python(timeout, int) log.debug1("zone.addRichRule('%s', '%s')" % (zone, rule)) obj = Rich_Rule(rule_str=rule) _zone = self.fw.zone.add_rule(zone, obj, timeout) if timeout > 0: tag = GLib.timeout_add_seconds(timeout, self.disableTimedRichRule, _zone, rule) self.addTimeout(_zone, rule, tag) self.RichRuleAdded(_zone, rule, timeout) return _zone @dbus_polkit_require_auth(config.dbus.PK_ACTION_CONFIG) @dbus_service_method(config.dbus.DBUS_INTERFACE_ZONE, in_signature='ss', out_signature='s') @dbus_handle_exceptions def removeRichRule(self, zone, rule, sender=None): # pylint: disable=W0613 zone = dbus_to_python(zone, str) rule = dbus_to_python(rule, str) log.debug1("zone.removeRichRule('%s', '%s')" % (zone, rule)) obj = Rich_Rule(rule_str=rule) _zone = self.fw.zone.remove_rule(zone, obj) self.removeTimeout(_zone, rule) self.RichRuleRemoved(_zone, rule) return _zone @dbus_polkit_require_auth(config.dbus.PK_ACTION_CONFIG_INFO) @dbus_service_method(config.dbus.DBUS_INTERFACE_ZONE, in_signature='ss', out_signature='b') @dbus_handle_exceptions def queryRichRule(self, zone, rule, sender=None): # pylint: disable=W0613 zone = dbus_to_python(zone, str) rule = dbus_to_python(rule, str) log.debug1("zone.queryRichRule('%s', '%s')" % (zone, rule)) obj = Rich_Rule(rule_str=rule) return self.fw.zone.query_rule(zone, obj) @dbus_polkit_require_auth(config.dbus.PK_ACTION_CONFIG_INFO) @dbus_service_method(config.dbus.DBUS_INTERFACE_ZONE, in_signature='s', out_signature='as') @dbus_handle_exceptions def getRichRules(self, zone, sender=None): # pylint: disable=W0613 # returns the list of enabled rich rules for zone # TODO: should be renamed to listRichRules() # because is called by firewall-cmd --zone --list-rich-rules zone = dbus_to_python(zone, str) log.debug1("zone.getRichRules('%s')" % (zone)) return self.fw.zone.list_rules(zone) @dbus.service.signal(config.dbus.DBUS_INTERFACE_ZONE, signature='ssi') @dbus_handle_exceptions def RichRuleAdded(self, zone, rule, timeout): log.debug1("zone.RichRuleAdded('%s', '%s', %d)" % (zone, rule, timeout)) @dbus.service.signal(config.dbus.DBUS_INTERFACE_ZONE, signature='ss') @dbus_handle_exceptions def RichRuleRemoved(self, zone, rule): log.debug1("zone.RichRuleRemoved('%s', '%s')" % (zone, rule)) # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # SERVICES @dbus_handle_exceptions def disableTimedService(self, zone, service): log.debug1("zone.disableTimedService('%s', '%s')" % (zone, service)) del self._timeouts[zone][service] self.fw.zone.remove_service(zone, service) self.ServiceRemoved(zone, service) @dbus_polkit_require_auth(config.dbus.PK_ACTION_CONFIG) @dbus_service_method(config.dbus.DBUS_INTERFACE_ZONE, in_signature='ssi', out_signature='s') @dbus_handle_exceptions def addService(self, zone, service, timeout, sender=None): # enables service <service> if not enabled already for zone zone = dbus_to_python(zone, str) service = dbus_to_python(service, str) timeout = dbus_to_python(timeout, int) log.debug1("zone.addService('%s', '%s', %d)" % (zone, service, timeout)) self.accessCheck(sender) _zone = self.fw.zone.add_service(zone, service, timeout, sender) if timeout > 0: tag = GLib.timeout_add_seconds(timeout, self.disableTimedService, _zone, service) self.addTimeout(_zone, service, tag) self.ServiceAdded(_zone, service, timeout) return _zone @dbus_polkit_require_auth(config.dbus.PK_ACTION_CONFIG) @dbus_service_method(config.dbus.DBUS_INTERFACE_ZONE, in_signature='ss', out_signature='s') @dbus_handle_exceptions def removeService(self, zone, service, sender=None): # disables service for zone zone = dbus_to_python(zone, str) service = dbus_to_python(service, str) log.debug1("zone.removeService('%s', '%s')" % (zone, service)) self.accessCheck(sender) _zone = self.fw.zone.remove_service(zone, service) self.removeTimeout(_zone, service) self.ServiceRemoved(_zone, service) return _zone @dbus_polkit_require_auth(config.dbus.PK_ACTION_CONFIG_INFO) @dbus_service_method(config.dbus.DBUS_INTERFACE_ZONE, in_signature='ss', out_signature='b') @dbus_handle_exceptions def queryService(self, zone, service, sender=None): # pylint: disable=W0613 # returns true if a service is enabled for zone zone = dbus_to_python(zone, str) service = dbus_to_python(service, str) log.debug1("zone.queryService('%s', '%s')" % (zone, service)) return self.fw.zone.query_service(zone, service) @dbus_polkit_require_auth(config.dbus.PK_ACTION_CONFIG_INFO) @dbus_service_method(config.dbus.DBUS_INTERFACE_ZONE, in_signature='s', out_signature='as') @dbus_handle_exceptions def getServices(self, zone, sender=None): # pylint: disable=W0613 # returns the list of enabled services for zone # TODO: should be renamed to listServices() # because is called by firewall-cmd --zone --list-services zone = dbus_to_python(zone, str) log.debug1("zone.getServices('%s')" % (zone)) return self.fw.zone.list_services(zone) @dbus.service.signal(config.dbus.DBUS_INTERFACE_ZONE, signature='ssi') @dbus_handle_exceptions def ServiceAdded(self, zone, service, timeout): log.debug1("zone.ServiceAdded('%s', '%s', %d)" % \ (zone, service, timeout)) @dbus.service.signal(config.dbus.DBUS_INTERFACE_ZONE, signature='ss') @dbus_handle_exceptions def ServiceRemoved(self, zone, service): log.debug1("zone.ServiceRemoved('%s', '%s')" % (zone, service)) # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # PORTS @dbus_handle_exceptions def disableTimedPort(self, zone, port, protocol): log.debug1("zone.disableTimedPort('%s', '%s', '%s')" % \ (zone, port, protocol)) del self._timeouts[zone][(port, protocol)] self.fw.zone.remove_port(zone, port, protocol) self.PortRemoved(zone, port, protocol) @dbus_polkit_require_auth(config.dbus.PK_ACTION_CONFIG) @dbus_service_method(config.dbus.DBUS_INTERFACE_ZONE, in_signature='sssi', out_signature='s') @dbus_handle_exceptions def addPort(self, zone, port, protocol, timeout, sender=None): # pylint: disable=R0913 # adds port <port> <protocol> if not enabled already to zone zone = dbus_to_python(zone, str) port = dbus_to_python(port, str) protocol = dbus_to_python(protocol, str) timeout = dbus_to_python(timeout, int) log.debug1("zone.addPort('%s', '%s', '%s')" % \ (zone, port, protocol)) self.accessCheck(sender) _zone = self.fw.zone.add_port(zone, port, protocol, timeout, sender) if timeout > 0: tag = GLib.timeout_add_seconds(timeout, self.disableTimedPort, _zone, port, protocol) self.addTimeout(_zone, (port, protocol), tag) self.PortAdded(_zone, port, protocol, timeout) return _zone @dbus_polkit_require_auth(config.dbus.PK_ACTION_CONFIG) @dbus_service_method(config.dbus.DBUS_INTERFACE_ZONE, in_signature='sss', out_signature='s') @dbus_handle_exceptions def removePort(self, zone, port, protocol, sender=None): # pylint: disable=R0913 # removes port<port> <protocol> if enabled from zone zone = dbus_to_python(zone, str) port = dbus_to_python(port, str) protocol = dbus_to_python(protocol, str) log.debug1("zone.removePort('%s', '%s', '%s')" % \ (zone, port, protocol)) self.accessCheck(sender) _zone= self.fw.zone.remove_port(zone, port, protocol) self.removeTimeout(_zone, (port, protocol)) self.PortRemoved(_zone, port, protocol) return _zone @dbus_polkit_require_auth(config.dbus.PK_ACTION_CONFIG_INFO) @dbus_service_method(config.dbus.DBUS_INTERFACE_ZONE, in_signature='sss', out_signature='b') @dbus_handle_exceptions def queryPort(self, zone, port, protocol, sender=None): # pylint: disable=W0613, R0913 # returns true if a port is enabled for zone zone = dbus_to_python(zone, str) port = dbus_to_python(port, str) protocol = dbus_to_python(protocol, str) log.debug1("zone.queryPort('%s', '%s', '%s')" % (zone, port, protocol)) return self.fw.zone.query_port(zone, port, protocol) @dbus_polkit_require_auth(config.dbus.PK_ACTION_CONFIG_INFO) @dbus_service_method(config.dbus.DBUS_INTERFACE_ZONE, in_signature='s', out_signature='aas') @dbus_handle_exceptions def getPorts(self, zone, sender=None): # pylint: disable=W0613 # returns the list of enabled ports # TODO: should be renamed to listPorts() # because is called by firewall-cmd --zone --list-ports zone = dbus_to_python(zone, str) log.debug1("zone.getPorts('%s')" % (zone)) return self.fw.zone.list_ports(zone) @dbus.service.signal(config.dbus.DBUS_INTERFACE_ZONE, signature='sssi') @dbus_handle_exceptions def PortAdded(self, zone, port, protocol, timeout=0): log.debug1("zone.PortAdded('%s', '%s', '%s', %d)" % \ (zone, port, protocol, timeout)) @dbus.service.signal(config.dbus.DBUS_INTERFACE_ZONE, signature='sss') @dbus_handle_exceptions def PortRemoved(self, zone, port, protocol): log.debug1("zone.PortRemoved('%s', '%s', '%s')" % \ (zone, port, protocol)) # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # PROTOCOLS @dbus_handle_exceptions def disableTimedProtocol(self, zone, protocol): log.debug1("zone.disableTimedProtocol('%s', '%s')" % (zone, protocol)) del self._timeouts[zone][(protocol)] self.fw.zone.remove_protocol(zone, protocol) self.ProtocolRemoved(zone, protocol) @dbus_polkit_require_auth(config.dbus.PK_ACTION_CONFIG) @dbus_service_method(config.dbus.DBUS_INTERFACE_ZONE, in_signature='ssi', out_signature='s') @dbus_handle_exceptions def addProtocol(self, zone, protocol, timeout, sender=None): # adds protocol <protocol> if not enabled already to zone zone = dbus_to_python(zone, str) protocol = dbus_to_python(protocol, str) timeout = dbus_to_python(timeout, int) log.debug1("zone.enableProtocol('%s', '%s')" % (zone, protocol)) self.accessCheck(sender) _zone = self.fw.zone.add_protocol(zone, protocol, timeout, sender) if timeout > 0: tag = GLib.timeout_add_seconds(timeout, self.disableTimedProtocol, _zone, protocol) self.addTimeout(_zone, protocol, tag) self.ProtocolAdded(_zone, protocol, timeout) return _zone @dbus_polkit_require_auth(config.dbus.PK_ACTION_CONFIG) @dbus_service_method(config.dbus.DBUS_INTERFACE_ZONE, in_signature='ss', out_signature='s') @dbus_handle_exceptions def removeProtocol(self, zone, protocol, sender=None): # removes protocol<protocol> if enabled from zone zone = dbus_to_python(zone, str) protocol = dbus_to_python(protocol, str) log.debug1("zone.removeProtocol('%s', '%s')" % (zone, protocol)) self.accessCheck(sender) _zone= self.fw.zone.remove_protocol(zone, protocol) self.removeTimeout(_zone, protocol) self.ProtocolRemoved(_zone, protocol) return _zone @dbus_polkit_require_auth(config.dbus.PK_ACTION_CONFIG_INFO) @dbus_service_method(config.dbus.DBUS_INTERFACE_ZONE, in_signature='ss', out_signature='b') @dbus_handle_exceptions def queryProtocol(self, zone, protocol, sender=None): # pylint: disable=W0613 # returns true if a protocol is enabled for zone zone = dbus_to_python(zone, str) protocol = dbus_to_python(protocol, str) log.debug1("zone.queryProtocol('%s', '%s')" % (zone, protocol)) return self.fw.zone.query_protocol(zone, protocol) @dbus_polkit_require_auth(config.dbus.PK_ACTION_CONFIG_INFO) @dbus_service_method(config.dbus.DBUS_INTERFACE_ZONE, in_signature='s', out_signature='as') @dbus_handle_exceptions def getProtocols(self, zone, sender=None): # pylint: disable=W0613 # returns the list of enabled protocols # TODO: should be renamed to listProtocols() # because is called by firewall-cmd --zone --list-protocols zone = dbus_to_python(zone, str) log.debug1("zone.getProtocols('%s')" % (zone)) return self.fw.zone.list_protocols(zone) @dbus.service.signal(config.dbus.DBUS_INTERFACE_ZONE, signature='ssi') @dbus_handle_exceptions def ProtocolAdded(self, zone, protocol, timeout=0): log.debug1("zone.ProtocolAdded('%s', '%s', %d)" % \ (zone, protocol, timeout)) @dbus.service.signal(config.dbus.DBUS_INTERFACE_ZONE, signature='ss') @dbus_handle_exceptions def ProtocolRemoved(self, zone, protocol): log.debug1("zone.ProtocolRemoved('%s', '%s')" % (zone, protocol)) # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # SOURCE PORTS @dbus_handle_exceptions def disableTimedSourcePort(self, zone, port, protocol): log.debug1("zone.disableTimedSourcePort('%s', '%s', '%s')" % \ (zone, port, protocol)) del self._timeouts[zone][("sport", port, protocol)] self.fw.zone.remove_source_port(zone, port, protocol) self.SourcePortRemoved(zone, port, protocol) @dbus_polkit_require_auth(config.dbus.PK_ACTION_CONFIG) @dbus_service_method(config.dbus.DBUS_INTERFACE_ZONE, in_signature='sssi', out_signature='s') @dbus_handle_exceptions def addSourcePort(self, zone, port, protocol, timeout, sender=None): # pylint: disable=R0913 # adds source port <port> <protocol> if not enabled already to zone zone = dbus_to_python(zone, str) port = dbus_to_python(port, str) protocol = dbus_to_python(protocol, str) timeout = dbus_to_python(timeout, int) log.debug1("zone.addSourcePort('%s', '%s', '%s')" % (zone, port, protocol)) self.accessCheck(sender) _zone = self.fw.zone.add_source_port(zone, port, protocol, timeout, sender) if timeout > 0: tag = GLib.timeout_add_seconds(timeout, self.disableTimedSourcePort, _zone, port, protocol) self.addTimeout(_zone, ("sport", port, protocol), tag) self.SourcePortAdded(_zone, port, protocol, timeout) return _zone @dbus_polkit_require_auth(config.dbus.PK_ACTION_CONFIG) @dbus_service_method(config.dbus.DBUS_INTERFACE_ZONE, in_signature='sss', out_signature='s') @dbus_handle_exceptions def removeSourcePort(self, zone, port, protocol, sender=None): # pylint: disable=R0913 # removes source port<port> <protocol> if enabled from zone zone = dbus_to_python(zone, str) port = dbus_to_python(port, str) protocol = dbus_to_python(protocol, str) log.debug1("zone.removeSourcePort('%s', '%s', '%s')" % (zone, port, protocol)) self.accessCheck(sender) _zone= self.fw.zone.remove_source_port(zone, port, protocol) self.removeTimeout(_zone, ("sport", port, protocol)) self.SourcePortRemoved(_zone, port, protocol) return _zone @dbus_polkit_require_auth(config.dbus.PK_ACTION_CONFIG_INFO) @dbus_service_method(config.dbus.DBUS_INTERFACE_ZONE, in_signature='sss', out_signature='b') @dbus_handle_exceptions def querySourcePort(self, zone, port, protocol, sender=None): # pylint: disable=W0613, R0913 # returns true if a source port is enabled for zone zone = dbus_to_python(zone, str) port = dbus_to_python(port, str) protocol = dbus_to_python(protocol, str) log.debug1("zone.querySourcePort('%s', '%s', '%s')" % (zone, port, protocol)) return self.fw.zone.query_source_port(zone, port, protocol) @dbus_polkit_require_auth(config.dbus.PK_ACTION_CONFIG_INFO) @dbus_service_method(config.dbus.DBUS_INTERFACE_ZONE, in_signature='s', out_signature='aas') @dbus_handle_exceptions def getSourcePorts(self, zone, sender=None): # pylint: disable=W0613 # returns the list of enabled source ports # TODO: should be renamed to listSourcePorts() # because is called by firewall-cmd --zone --list-source-ports zone = dbus_to_python(zone, str) log.debug1("zone.getSourcePorts('%s')" % (zone)) return self.fw.zone.list_source_ports(zone) @dbus.service.signal(config.dbus.DBUS_INTERFACE_ZONE, signature='sssi') @dbus_handle_exceptions def SourcePortAdded(self, zone, port, protocol, timeout=0): log.debug1("zone.SourcePortAdded('%s', '%s', '%s', %d)" % \ (zone, port, protocol, timeout)) @dbus.service.signal(config.dbus.DBUS_INTERFACE_ZONE, signature='sss') @dbus_handle_exceptions def SourcePortRemoved(self, zone, port, protocol): log.debug1("zone.SourcePortRemoved('%s', '%s', '%s')" % (zone, port, protocol)) # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # MASQUERADE @dbus_handle_exceptions def disableTimedMasquerade(self, zone): del self._timeouts[zone]["masquerade"] self.fw.zone.remove_masquerade(zone) self.MasqueradeRemoved(zone) @dbus_polkit_require_auth(config.dbus.PK_ACTION_CONFIG) @dbus_service_method(config.dbus.DBUS_INTERFACE_ZONE, in_signature='si', out_signature='s') @dbus_handle_exceptions def addMasquerade(self, zone, timeout, sender=None): # adds masquerade if not added already zone = dbus_to_python(zone, str) timeout = dbus_to_python(timeout, int) log.debug1("zone.addMasquerade('%s')" % (zone)) self.accessCheck(sender) _zone = self.fw.zone.add_masquerade(zone, timeout, sender) if timeout > 0: tag = GLib.timeout_add_seconds(timeout, self.disableTimedMasquerade, _zone) self.addTimeout(_zone, "masquerade", tag) self.MasqueradeAdded(_zone, timeout) return _zone @dbus_polkit_require_auth(config.dbus.PK_ACTION_CONFIG) @dbus_service_method(config.dbus.DBUS_INTERFACE_ZONE, in_signature='s', out_signature='s') @dbus_handle_exceptions def removeMasquerade(self, zone, sender=None): # removes masquerade zone = dbus_to_python(zone, str) log.debug1("zone.removeMasquerade('%s')" % (zone)) self.accessCheck(sender) _zone = self.fw.zone.remove_masquerade(zone) self.removeTimeout(_zone, "masquerade") self.MasqueradeRemoved(_zone) return _zone @dbus_polkit_require_auth(config.dbus.PK_ACTION_CONFIG_INFO) @dbus_service_method(config.dbus.DBUS_INTERFACE_ZONE, in_signature='s', out_signature='b') @dbus_handle_exceptions def queryMasquerade(self, zone, sender=None): # pylint: disable=W0613 # returns true if a masquerade is added zone = dbus_to_python(zone, str) log.debug1("zone.queryMasquerade('%s')" % (zone)) return self.fw.zone.query_masquerade(zone) @dbus.service.signal(config.dbus.DBUS_INTERFACE_ZONE, signature='si') @dbus_handle_exceptions def MasqueradeAdded(self, zone, timeout=0): log.debug1("zone.MasqueradeAdded('%s', %d)" % (zone, timeout)) @dbus.service.signal(config.dbus.DBUS_INTERFACE_ZONE, signature='s') @dbus_handle_exceptions def MasqueradeRemoved(self, zone): log.debug1("zone.MasqueradeRemoved('%s')" % (zone)) # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # FORWARD PORT @dbus_handle_exceptions def disable_forward_port(self, zone, port, protocol, toport, toaddr): # pylint: disable=R0913 del self._timeouts[zone][(port, protocol, toport, toaddr)] self.fw.zone.remove_forward_port(zone, port, protocol, toport, toaddr) self.ForwardPortRemoved(zone, port, protocol, toport, toaddr) @dbus_polkit_require_auth(config.dbus.PK_ACTION_CONFIG) @dbus_service_method(config.dbus.DBUS_INTERFACE_ZONE, in_signature='sssssi', out_signature='s') @dbus_handle_exceptions def addForwardPort(self, zone, port, protocol, toport, toaddr, timeout, sender=None): # pylint: disable=R0913 # add forward port if not enabled already for zone zone = dbus_to_python(zone, str) port = dbus_to_python(port, str) protocol = dbus_to_python(protocol, str) toport = dbus_to_python(toport, str) toaddr = dbus_to_python(toaddr, str) timeout = dbus_to_python(timeout, int) log.debug1("zone.addForwardPort('%s', '%s', '%s', '%s', '%s')" % \ (zone, port, protocol, toport, toaddr)) self.accessCheck(sender) _zone = self.fw.zone.add_forward_port(zone, port, protocol, toport, toaddr, timeout, sender) if timeout > 0: tag = GLib.timeout_add_seconds(timeout, self.disable_forward_port, _zone, port, protocol, toport, toaddr) self.addTimeout(_zone, (port, protocol, toport, toaddr), tag) self.ForwardPortAdded(_zone, port, protocol, toport, toaddr, timeout) return _zone @dbus_polkit_require_auth(config.dbus.PK_ACTION_CONFIG) @dbus_service_method(config.dbus.DBUS_INTERFACE_ZONE, in_signature='sssss', out_signature='s') @dbus_handle_exceptions def removeForwardPort(self, zone, port, protocol, toport, toaddr, sender=None): # pylint: disable=R0913 # remove forward port from zone zone = dbus_to_python(zone, str) port = dbus_to_python(port, str) protocol = dbus_to_python(protocol, str) toport = dbus_to_python(toport, str) toaddr = dbus_to_python(toaddr, str) log.debug1("zone.removeForwardPort('%s', '%s', '%s', '%s', '%s')" % \ (zone, port, protocol, toport, toaddr)) self.accessCheck(sender) _zone = self.fw.zone.remove_forward_port(zone, port, protocol, toport, toaddr) self.removeTimeout(_zone, (port, protocol, toport, toaddr)) self.ForwardPortRemoved(_zone, port, protocol, toport, toaddr) return _zone @dbus_polkit_require_auth(config.dbus.PK_ACTION_CONFIG_INFO) @dbus_service_method(config.dbus.DBUS_INTERFACE_ZONE, in_signature='sssss', out_signature='b') @dbus_handle_exceptions def queryForwardPort(self, zone, port, protocol, toport, toaddr, sender=None): # pylint: disable=W0613, R0913 # returns true if a forward port is enabled for zone zone = dbus_to_python(zone, str) port = dbus_to_python(port, str) protocol = dbus_to_python(protocol, str) toport = dbus_to_python(toport, str) toaddr = dbus_to_python(toaddr, str) log.debug1("zone.queryForwardPort('%s', '%s', '%s', '%s', '%s')" % \ (zone, port, protocol, toport, toaddr)) return self.fw.zone.query_forward_port(zone, port, protocol, toport, toaddr) @dbus_polkit_require_auth(config.dbus.PK_ACTION_CONFIG_INFO) @dbus_service_method(config.dbus.DBUS_INTERFACE_ZONE, in_signature='s', out_signature='aas') @dbus_handle_exceptions def getForwardPorts(self, zone, sender=None): # pylint: disable=W0613 # returns the list of enabled ports for zone # TODO: should be renamed to listForwardPorts() # because is called by firewall-cmd --zone --list-forward-ports zone = dbus_to_python(zone, str) log.debug1("zone.getForwardPorts('%s')" % (zone)) return self.fw.zone.list_forward_ports(zone) @dbus.service.signal(config.dbus.DBUS_INTERFACE_ZONE, signature='sssssi') @dbus_handle_exceptions def ForwardPortAdded(self, zone, port, protocol, toport, toaddr, timeout=0): # pylint: disable=R0913 log.debug1("zone.ForwardPortAdded('%s', '%s', '%s', '%s', '%s', %d)" % \ (zone, port, protocol, toport, toaddr, timeout)) @dbus.service.signal(config.dbus.DBUS_INTERFACE_ZONE, signature='sssss') @dbus_handle_exceptions def ForwardPortRemoved(self, zone, port, protocol, toport, toaddr): # pylint: disable=R0913 log.debug1("zone.ForwardPortRemoved('%s', '%s', '%s', '%s', '%s')" % \ (zone, port, protocol, toport, toaddr)) # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # ICMP BLOCK @dbus_handle_exceptions def disableTimedIcmpBlock(self, zone, icmp, sender): # pylint: disable=W0613 log.debug1("zone.disableTimedIcmpBlock('%s', '%s')" % (zone, icmp)) del self._timeouts[zone][icmp] self.fw.zone.remove_icmp_block(zone, icmp) self.IcmpBlockRemoved(zone, icmp) @dbus_polkit_require_auth(config.dbus.PK_ACTION_CONFIG) @dbus_service_method(config.dbus.DBUS_INTERFACE_ZONE, in_signature='ssi', out_signature='s') @dbus_handle_exceptions def addIcmpBlock(self, zone, icmp, timeout, sender=None): # add icmpblock <icmp> if not enabled already for zone zone = dbus_to_python(zone, str) icmp = dbus_to_python(icmp, str) timeout = dbus_to_python(timeout, int) log.debug1("zone.enableIcmpBlock('%s', '%s')" % (zone, icmp)) self.accessCheck(sender) _zone = self.fw.zone.add_icmp_block(zone, icmp, timeout, sender) if timeout > 0: tag = GLib.timeout_add_seconds(timeout, self.disableTimedIcmpBlock, _zone, icmp, sender) self.addTimeout(_zone, icmp, tag) self.IcmpBlockAdded(_zone, icmp, timeout) return _zone @dbus_polkit_require_auth(config.dbus.PK_ACTION_CONFIG) @dbus_service_method(config.dbus.DBUS_INTERFACE_ZONE, in_signature='ss', out_signature='s') @dbus_handle_exceptions def removeIcmpBlock(self, zone, icmp, sender=None): # removes icmpBlock from zone zone = dbus_to_python(zone, str) icmp = dbus_to_python(icmp, str) log.debug1("zone.removeIcmpBlock('%s', '%s')" % (zone, icmp)) self.accessCheck(sender) _zone = self.fw.zone.remove_icmp_block(zone, icmp) self.removeTimeout(_zone, icmp) self.IcmpBlockRemoved(_zone, icmp) return _zone @dbus_polkit_require_auth(config.dbus.PK_ACTION_CONFIG_INFO) @dbus_service_method(config.dbus.DBUS_INTERFACE_ZONE, in_signature='ss', out_signature='b') @dbus_handle_exceptions def queryIcmpBlock(self, zone, icmp, sender=None): # pylint: disable=W0613 # returns true if a icmp is enabled for zone zone = dbus_to_python(zone, str) icmp = dbus_to_python(icmp, str) log.debug1("zone.queryIcmpBlock('%s', '%s')" % (zone, icmp)) return self.fw.zone.query_icmp_block(zone, icmp) @dbus_polkit_require_auth(config.dbus.PK_ACTION_CONFIG_INFO) @dbus_service_method(config.dbus.DBUS_INTERFACE_ZONE, in_signature='s', out_signature='as') @dbus_handle_exceptions def getIcmpBlocks(self, zone, sender=None): # pylint: disable=W0613 # returns the list of enabled icmpblocks # TODO: should be renamed to listIcmpBlocks() # because is called by firewall-cmd --zone --list-icmp-blocks zone = dbus_to_python(zone, str) log.debug1("zone.getIcmpBlocks('%s')" % (zone)) return self.fw.zone.list_icmp_blocks(zone) @dbus.service.signal(config.dbus.DBUS_INTERFACE_ZONE, signature='ssi') @dbus_handle_exceptions def IcmpBlockAdded(self, zone, icmp, timeout=0): log.debug1("zone.IcmpBlockAdded('%s', '%s', %d)" % \ (zone, icmp, timeout)) @dbus.service.signal(config.dbus.DBUS_INTERFACE_ZONE, signature='ss') @dbus_handle_exceptions def IcmpBlockRemoved(self, zone, icmp): log.debug1("zone.IcmpBlockRemoved('%s', '%s')" % (zone, icmp)) # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # ICMP BLOCK INVERSION @dbus_polkit_require_auth(config.dbus.PK_ACTION_CONFIG) @dbus_service_method(config.dbus.DBUS_INTERFACE_ZONE, in_signature='s', out_signature='s') @dbus_handle_exceptions def addIcmpBlockInversion(self, zone, sender=None): # adds icmpBlockInversion if not added already zone = dbus_to_python(zone, str) log.debug1("zone.addIcmpBlockInversion('%s')" % (zone)) self.accessCheck(sender) _zone = self.fw.zone.add_icmp_block_inversion(zone, sender) self.IcmpBlockInversionAdded(_zone) return _zone @dbus_polkit_require_auth(config.dbus.PK_ACTION_CONFIG) @dbus_service_method(config.dbus.DBUS_INTERFACE_ZONE, in_signature='s', out_signature='s') @dbus_handle_exceptions def removeIcmpBlockInversion(self, zone, sender=None): # removes icmpBlockInversion zone = dbus_to_python(zone, str) log.debug1("zone.removeIcmpBlockInversion('%s')" % (zone)) self.accessCheck(sender) _zone = self.fw.zone.remove_icmp_block_inversion(zone) self.IcmpBlockInversionRemoved(_zone) return _zone @dbus_polkit_require_auth(config.dbus.PK_ACTION_CONFIG_INFO) @dbus_service_method(config.dbus.DBUS_INTERFACE_ZONE, in_signature='s', out_signature='b') @dbus_handle_exceptions def queryIcmpBlockInversion(self, zone, sender=None): # pylint: disable=W0613 # returns true if a icmpBlockInversion is added zone = dbus_to_python(zone, str) log.debug1("zone.queryIcmpBlockInversion('%s')" % (zone)) return self.fw.zone.query_icmp_block_inversion(zone) @dbus.service.signal(config.dbus.DBUS_INTERFACE_ZONE, signature='s') @dbus_handle_exceptions def IcmpBlockInversionAdded(self, zone): log.debug1("zone.IcmpBlockInversionAdded('%s')" % (zone)) @dbus.service.signal(config.dbus.DBUS_INTERFACE_ZONE, signature='s') @dbus_handle_exceptions def IcmpBlockInversionRemoved(self, zone): log.debug1("zone.IcmpBlockInversionRemoved('%s')" % (zone)) # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # DIRECT INTERFACE # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # DIRECT CHAIN @dbus_polkit_require_auth(config.dbus.PK_ACTION_DIRECT) @dbus_service_method_deprecated(config.dbus.DBUS_INTERFACE_DIRECT) @dbus_service_method(config.dbus.DBUS_INTERFACE_DIRECT, in_signature='sss', out_signature='') @dbus_handle_exceptions def addChain(self, ipv, table, chain, sender=None): # inserts direct chain ipv = dbus_to_python(ipv, str) table = dbus_to_python(table, str) chain = dbus_to_python(chain, str) log.debug1("direct.addChain('%s', '%s', '%s')" % (ipv, table, chain)) self.accessCheck(sender) self.fw.direct.add_chain(ipv, table, chain) self.ChainAdded(ipv, table, chain) @dbus_polkit_require_auth(config.dbus.PK_ACTION_DIRECT) @dbus_service_method_deprecated(config.dbus.DBUS_INTERFACE_DIRECT) @dbus_service_method(config.dbus.DBUS_INTERFACE_DIRECT, in_signature='sss', out_signature='') @dbus_handle_exceptions def removeChain(self, ipv, table, chain, sender=None): # removes direct chain ipv = dbus_to_python(ipv, str) table = dbus_to_python(table, str) chain = dbus_to_python(chain, str) log.debug1("direct.removeChain('%s', '%s', '%s')" % (ipv, table, chain)) self.accessCheck(sender) self.fw.direct.remove_chain(ipv, table, chain) self.ChainRemoved(ipv, table, chain) @dbus_polkit_require_auth(config.dbus.PK_ACTION_DIRECT_INFO) @dbus_service_method_deprecated(config.dbus.DBUS_INTERFACE_DIRECT) @dbus_service_method(config.dbus.DBUS_INTERFACE_DIRECT, in_signature='sss', out_signature='b') @dbus_handle_exceptions def queryChain(self, ipv, table, chain, sender=None): # pylint: disable=W0613 # returns true if a chain is enabled ipv = dbus_to_python(ipv, str) table = dbus_to_python(table, str) chain = dbus_to_python(chain, str) log.debug1("direct.queryChain('%s', '%s', '%s')" % (ipv, table, chain)) return self.fw.direct.query_chain(ipv, table, chain) @dbus_polkit_require_auth(config.dbus.PK_ACTION_DIRECT_INFO) @dbus_service_method_deprecated(config.dbus.DBUS_INTERFACE_DIRECT) @dbus_service_method(config.dbus.DBUS_INTERFACE_DIRECT, in_signature='ss', out_signature='as') @dbus_handle_exceptions def getChains(self, ipv, table, sender=None): # pylint: disable=W0613 # returns list of added chains ipv = dbus_to_python(ipv, str) table = dbus_to_python(table, str) log.debug1("direct.getChains('%s', '%s')" % (ipv, table)) return self.fw.direct.get_chains(ipv, table) @dbus_polkit_require_auth(config.dbus.PK_ACTION_DIRECT_INFO) @dbus_service_method_deprecated(config.dbus.DBUS_INTERFACE_DIRECT) @dbus_service_method(config.dbus.DBUS_INTERFACE_DIRECT, in_signature='', out_signature='a(sss)') @dbus_handle_exceptions def getAllChains(self, sender=None): # pylint: disable=W0613 # returns list of added chains log.debug1("direct.getAllChains()") return self.fw.direct.get_all_chains() @dbus_service_signal_deprecated(config.dbus.DBUS_INTERFACE_DIRECT) @dbus.service.signal(config.dbus.DBUS_INTERFACE_DIRECT, signature='sss') @dbus_handle_exceptions def ChainAdded(self, ipv, table, chain): log.debug1("direct.ChainAdded('%s', '%s', '%s')" % (ipv, table, chain)) @dbus_service_signal_deprecated(config.dbus.DBUS_INTERFACE_DIRECT) @dbus.service.signal(config.dbus.DBUS_INTERFACE_DIRECT, signature='sss') @dbus_handle_exceptions def ChainRemoved(self, ipv, table, chain): log.debug1("direct.ChainRemoved('%s', '%s', '%s')" % (ipv, table, chain)) # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # DIRECT RULE @dbus_polkit_require_auth(config.dbus.PK_ACTION_DIRECT) @dbus_service_method_deprecated(config.dbus.DBUS_INTERFACE_DIRECT) @dbus_service_method(config.dbus.DBUS_INTERFACE_DIRECT, in_signature='sssias', out_signature='') @dbus_handle_exceptions def addRule(self, ipv, table, chain, priority, args, sender=None): # pylint: disable=R0913 # inserts direct rule ipv = dbus_to_python(ipv, str) table = dbus_to_python(table, str) chain = dbus_to_python(chain, str) priority = dbus_to_python(priority, int) args = tuple( dbus_to_python(i, str) for i in args ) log.debug1("direct.addRule('%s', '%s', '%s', %d, '%s')" % \ (ipv, table, chain, priority, "','".join(args))) self.accessCheck(sender) self.fw.direct.add_rule(ipv, table, chain, priority, args) self.RuleAdded(ipv, table, chain, priority, args) @dbus_polkit_require_auth(config.dbus.PK_ACTION_DIRECT) @dbus_service_method_deprecated(config.dbus.DBUS_INTERFACE_DIRECT) @dbus_service_method(config.dbus.DBUS_INTERFACE_DIRECT, in_signature='sssias', out_signature='') @dbus_handle_exceptions def removeRule(self, ipv, table, chain, priority, args, sender=None): # pylint: disable=R0913 # removes direct rule ipv = dbus_to_python(ipv, str) table = dbus_to_python(table, str) chain = dbus_to_python(chain, str) priority = dbus_to_python(priority, int) args = tuple( dbus_to_python(i, str) for i in args ) log.debug1("direct.removeRule('%s', '%s', '%s', %d, '%s')" % \ (ipv, table, chain, priority, "','".join(args))) self.accessCheck(sender) self.fw.direct.remove_rule(ipv, table, chain, priority, args) self.RuleRemoved(ipv, table, chain, priority, args) @dbus_polkit_require_auth(config.dbus.PK_ACTION_DIRECT) @dbus_service_method_deprecated(config.dbus.DBUS_INTERFACE_DIRECT) @dbus_service_method(config.dbus.DBUS_INTERFACE_DIRECT, in_signature='sss', out_signature='') @dbus_handle_exceptions def removeRules(self, ipv, table, chain, sender=None): # removes direct rule ipv = dbus_to_python(ipv, str) table = dbus_to_python(table, str) chain = dbus_to_python(chain, str) log.debug1("direct.removeRules('%s', '%s', '%s')" % (ipv, table, chain)) self.accessCheck(sender) for (priority, args) in self.fw.direct.get_rules(ipv, table, chain): self.fw.direct.remove_rule(ipv, table, chain, priority, args) self.RuleRemoved(ipv, table, chain, priority, args) @dbus_polkit_require_auth(config.dbus.PK_ACTION_DIRECT_INFO) @dbus_service_method_deprecated(config.dbus.DBUS_INTERFACE_DIRECT) @dbus_service_method(config.dbus.DBUS_INTERFACE_DIRECT, in_signature='sssias', out_signature='b') @dbus_handle_exceptions def queryRule(self, ipv, table, chain, priority, args, sender=None): # pylint: disable=W0613, R0913 # returns true if a rule is enabled ipv = dbus_to_python(ipv, str) table = dbus_to_python(table, str) chain = dbus_to_python(chain, str) priority = dbus_to_python(priority, int) args = tuple( dbus_to_python(i, str) for i in args ) log.debug1("direct.queryRule('%s', '%s', '%s', %d, '%s')" % \ (ipv, table, chain, priority, "','".join(args))) return self.fw.direct.query_rule(ipv, table, chain, priority, args) @dbus_polkit_require_auth(config.dbus.PK_ACTION_DIRECT_INFO) @dbus_service_method_deprecated(config.dbus.DBUS_INTERFACE_DIRECT) @dbus_service_method(config.dbus.DBUS_INTERFACE_DIRECT, in_signature='sss', out_signature='a(ias)') @dbus_handle_exceptions def getRules(self, ipv, table, chain, sender=None): # pylint: disable=W0613 # returns list of added rules ipv = dbus_to_python(ipv, str) table = dbus_to_python(table, str) chain = dbus_to_python(chain, str) log.debug1("direct.getRules('%s', '%s', '%s')" % (ipv, table, chain)) return self.fw.direct.get_rules(ipv, table, chain) @dbus_polkit_require_auth(config.dbus.PK_ACTION_DIRECT_INFO) @dbus_service_method_deprecated(config.dbus.DBUS_INTERFACE_DIRECT) @dbus_service_method(config.dbus.DBUS_INTERFACE_DIRECT, in_signature='', out_signature='a(sssias)') @dbus_handle_exceptions def getAllRules(self, sender=None): # pylint: disable=W0613 # returns list of added rules log.debug1("direct.getAllRules()") return self.fw.direct.get_all_rules() @dbus_service_signal_deprecated(config.dbus.DBUS_INTERFACE_DIRECT) @dbus.service.signal(config.dbus.DBUS_INTERFACE_DIRECT, signature='sssias') @dbus_handle_exceptions def RuleAdded(self, ipv, table, chain, priority, args): # pylint: disable=R0913 log.debug1("direct.RuleAdded('%s', '%s', '%s', %d, '%s')" % \ (ipv, table, chain, priority, "','".join(args))) @dbus_service_signal_deprecated(config.dbus.DBUS_INTERFACE_DIRECT) @dbus.service.signal(config.dbus.DBUS_INTERFACE_DIRECT, signature='sssias') @dbus_handle_exceptions def RuleRemoved(self, ipv, table, chain, priority, args): # pylint: disable=R0913 log.debug1("direct.RuleRemoved('%s', '%s', '%s', %d, '%s')" % \ (ipv, table, chain, priority, "','".join(args))) # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # DIRECT PASSTHROUGH (untracked) @dbus_polkit_require_auth(config.dbus.PK_ACTION_DIRECT) @dbus_service_method_deprecated(config.dbus.DBUS_INTERFACE_DIRECT) @dbus_service_method(config.dbus.DBUS_INTERFACE_DIRECT, in_signature='sas', out_signature='s') @dbus_handle_exceptions def passthrough(self, ipv, args, sender=None): # inserts direct rule ipv = dbus_to_python(ipv, str) args = tuple( dbus_to_python(i, str) for i in args ) log.debug1("direct.passthrough('%s', '%s')" % (ipv, "','".join(args))) self.accessCheck(sender) try: return self.fw.direct.passthrough(ipv, args) except FirewallError as error: if ipv in ["ipv4", "ipv6"]: query_args = set(["-C", "--check", "-L", "--list"]) else: query_args = set(["-L", "--list"]) msg = str(error) if error.code == errors.COMMAND_FAILED: if len(set(args) & query_args) <= 0: log.warning(msg) raise FirewallDBusException(msg) raise # DIRECT PASSTHROUGH (tracked) @dbus_polkit_require_auth(config.dbus.PK_ACTION_DIRECT) @dbus_service_method_deprecated(config.dbus.DBUS_INTERFACE_DIRECT) @dbus_service_method(config.dbus.DBUS_INTERFACE_DIRECT, in_signature='sas', out_signature='') @dbus_handle_exceptions def addPassthrough(self, ipv, args, sender=None): # inserts direct passthrough ipv = dbus_to_python(ipv) args = tuple( dbus_to_python(i) for i in args ) log.debug1("direct.addPassthrough('%s', '%s')" % \ (ipv, "','".join(args))) self.accessCheck(sender) self.fw.direct.add_passthrough(ipv, args) self.PassthroughAdded(ipv, args) @dbus_polkit_require_auth(config.dbus.PK_ACTION_DIRECT) @dbus_service_method_deprecated(config.dbus.DBUS_INTERFACE_DIRECT) @dbus_service_method(config.dbus.DBUS_INTERFACE_DIRECT, in_signature='sas', out_signature='') @dbus_handle_exceptions def removePassthrough(self, ipv, args, sender=None): # removes direct passthrough ipv = dbus_to_python(ipv) args = tuple( dbus_to_python(i) for i in args ) log.debug1("direct.removePassthrough('%s', '%s')" % \ (ipv, "','".join(args))) self.accessCheck(sender) self.fw.direct.remove_passthrough(ipv, args) self.PassthroughRemoved(ipv, args) @dbus_polkit_require_auth(config.dbus.PK_ACTION_DIRECT_INFO) @dbus_service_method_deprecated(config.dbus.DBUS_INTERFACE_DIRECT) @dbus_service_method(config.dbus.DBUS_INTERFACE_DIRECT, in_signature='sas', out_signature='b') @dbus_handle_exceptions def queryPassthrough(self, ipv, args, sender=None): # pylint: disable=W0613 # returns true if a passthrough is enabled ipv = dbus_to_python(ipv) args = tuple( dbus_to_python(i) for i in args ) log.debug1("direct.queryPassthrough('%s', '%s')" % \ (ipv, "','".join(args))) return self.fw.direct.query_passthrough(ipv, args) @dbus_polkit_require_auth(config.dbus.PK_ACTION_DIRECT_INFO) @dbus_service_method_deprecated(config.dbus.DBUS_INTERFACE_DIRECT) @dbus_service_method(config.dbus.DBUS_INTERFACE_DIRECT, in_signature='', out_signature='a(sas)') @dbus_handle_exceptions def getAllPassthroughs(self, sender=None): # pylint: disable=W0613 # returns list of all added passthroughs log.debug1("direct.getAllPassthroughs()") return self.fw.direct.get_all_passthroughs() @dbus_polkit_require_auth(config.dbus.PK_ACTION_DIRECT) @dbus_service_method_deprecated(config.dbus.DBUS_INTERFACE_DIRECT) @dbus_service_method(config.dbus.DBUS_INTERFACE_DIRECT, in_signature='', out_signature='') @dbus_handle_exceptions def removeAllPassthroughs(self, sender=None): # pylint: disable=W0613 # remove all passhroughs log.debug1("direct.removeAllPassthroughs()") # remove in reverse order to avoid removing non-empty chains for passthrough in reversed(self.getAllPassthroughs()): self.removePassthrough(*passthrough) @dbus_polkit_require_auth(config.dbus.PK_ACTION_DIRECT_INFO) @dbus_service_method_deprecated(config.dbus.DBUS_INTERFACE_DIRECT) @dbus_service_method(config.dbus.DBUS_INTERFACE_DIRECT, in_signature='s', out_signature='aas') @dbus_handle_exceptions def getPassthroughs(self, ipv, sender=None): # pylint: disable=W0613 # returns list of all added passthroughs with ipv ipv = dbus_to_python(ipv) log.debug1("direct.getPassthroughs('%s')", ipv) return self.fw.direct.get_passthroughs(ipv) @dbus_service_signal_deprecated(config.dbus.DBUS_INTERFACE_DIRECT) @dbus.service.signal(config.dbus.DBUS_INTERFACE_DIRECT, signature='sas') @dbus_handle_exceptions def PassthroughAdded(self, ipv, args): log.debug1("direct.PassthroughAdded('%s', '%s')" % \ (ipv, "','".join(args))) @dbus_service_signal_deprecated(config.dbus.DBUS_INTERFACE_DIRECT) @dbus.service.signal(config.dbus.DBUS_INTERFACE_DIRECT, signature='sas') @dbus_handle_exceptions def PassthroughRemoved(self, ipv, args): log.debug1("direct.PassthroughRemoved('%s', '%s')" % \ (ipv, "','".join(args))) # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # @dbus_polkit_require_auth(config.dbus.PK_ACTION_ALL) @dbus_service_method(config.dbus.DBUS_INTERFACE, in_signature='', out_signature='') @dbus_handle_exceptions def authorizeAll(self, sender=None): # pylint: disable=W0613 """ PK_ACTION_ALL implies all other actions, i.e. once a subject is authorized for PK_ACTION_ALL it's also authorized for any other action. Use-case is GUI (RHBZ#994729). """ pass # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # IPSETS # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # @dbus_polkit_require_auth(config.dbus.PK_ACTION_INFO) @dbus_service_method(config.dbus.DBUS_INTERFACE_IPSET, in_signature='s', out_signature='b') @dbus_handle_exceptions def queryIPSet(self, ipset, sender=None): # pylint: disable=W0613 # returns true if a set with the name exists ipset = dbus_to_python(ipset) log.debug1("ipset.queryIPSet('%s')" % (ipset)) return self.fw.ipset.query_ipset(ipset) @dbus_polkit_require_auth(config.dbus.PK_ACTION_INFO) @dbus_service_method(config.dbus.DBUS_INTERFACE_IPSET, in_signature='', out_signature='as') @dbus_handle_exceptions def getIPSets(self, sender=None): # pylint: disable=W0613 # returns list of added sets log.debug1("ipsets.getIPSets()") return self.fw.ipset.get_ipsets() @dbus_polkit_require_auth(config.dbus.PK_ACTION_CONFIG_INFO) @dbus_service_method(config.dbus.DBUS_INTERFACE_IPSET, in_signature='s', out_signature=IPSet.DBUS_SIGNATURE) @dbus_handle_exceptions def getIPSetSettings(self, ipset, sender=None): # pylint: disable=W0613 # returns ipset settings for ipset ipset = dbus_to_python(ipset, str) log.debug1("getIPSetSettings(%s)", ipset) return self.fw.ipset.get_ipset(ipset).export_config() # set entries # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # @dbus_polkit_require_auth(config.dbus.PK_ACTION_CONFIG) @dbus_service_method(config.dbus.DBUS_INTERFACE_IPSET, in_signature='ss', out_signature='') @dbus_handle_exceptions def addEntry(self, ipset, entry, sender=None): # adds ipset entry ipset = dbus_to_python(ipset) entry = dbus_to_python(entry) log.debug1("ipset.addEntry('%s', '%s')" % (ipset, entry)) self.accessCheck(sender) self.fw.ipset.add_entry(ipset, entry) self.EntryAdded(ipset, entry) @dbus_polkit_require_auth(config.dbus.PK_ACTION_CONFIG) @dbus_service_method(config.dbus.DBUS_INTERFACE_IPSET, in_signature='ss', out_signature='') @dbus_handle_exceptions def removeEntry(self, ipset, entry, sender=None): # removes ipset entry ipset = dbus_to_python(ipset) entry = dbus_to_python(entry) log.debug1("ipset.removeEntry('%s', '%s')" % (ipset, entry)) self.accessCheck(sender) self.fw.ipset.remove_entry(ipset, entry) self.EntryRemoved(ipset, entry) @dbus_polkit_require_auth(config.dbus.PK_ACTION_INFO) @dbus_service_method(config.dbus.DBUS_INTERFACE_IPSET, in_signature='ss', out_signature='b') @dbus_handle_exceptions def queryEntry(self, ipset, entry, sender=None): # pylint: disable=W0613 # returns true if the entry exists in the ipset ipset = dbus_to_python(ipset) entry = dbus_to_python(entry) log.debug1("ipset.queryEntry('%s', '%s')" % (ipset, entry)) return self.fw.ipset.query_entry(ipset, entry) @dbus_polkit_require_auth(config.dbus.PK_ACTION_INFO) @dbus_service_method(config.dbus.DBUS_INTERFACE_IPSET, in_signature='s', out_signature='as') @dbus_handle_exceptions def getEntries(self, ipset, sender=None): # pylint: disable=W0613 # returns list of added entries for the ipset ipset = dbus_to_python(ipset) log.debug1("ipset.getEntries('%s')" % ipset) return self.fw.ipset.get_entries(ipset) @dbus_polkit_require_auth(config.dbus.PK_ACTION_CONFIG) @dbus_service_method(config.dbus.DBUS_INTERFACE_IPSET, in_signature='sas') @dbus_handle_exceptions def setEntries(self, ipset, entries, sender=None): # pylint: disable=W0613 # returns list of added entries for the ipset ipset = dbus_to_python(ipset) entries = dbus_to_python(entries, list) log.debug1("ipset.setEntries('%s', '[%s]')", ipset, ",".join(entries)) old_entries = self.fw.ipset.get_entries(ipset) self.fw.ipset.set_entries(ipset, entries) old_entries_set = set(old_entries) entries_set = set(entries) for entry in entries_set - old_entries_set: self.EntryAdded(ipset, entry) for entry in old_entries_set - entries_set: self.EntryRemoved(ipset, entry) @dbus.service.signal(config.dbus.DBUS_INTERFACE_IPSET, signature='ss') @dbus_handle_exceptions def EntryAdded(self, ipset, entry): ipset = dbus_to_python(ipset) entry = dbus_to_python(entry) log.debug1("ipset.EntryAdded('%s', '%s')" % (ipset, entry)) @dbus.service.signal(config.dbus.DBUS_INTERFACE_IPSET, signature='ss') @dbus_handle_exceptions def EntryRemoved(self, ipset, entry): ipset = dbus_to_python(ipset) entry = dbus_to_python(entry) log.debug1("ipset.EntryRemoved('%s', '%s')" % (ipset, entry)) # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # HELPERS # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # @dbus_polkit_require_auth(config.dbus.PK_ACTION_INFO) @dbus_service_method(config.dbus.DBUS_INTERFACE, in_signature='', out_signature='as') @dbus_handle_exceptions def getHelpers(self, sender=None): # pylint: disable=W0613 # returns list of added sets log.debug1("helpers.getHelpers()") return self.fw.helper.get_helpers() @dbus_polkit_require_auth(config.dbus.PK_ACTION_CONFIG_INFO) @dbus_service_method(config.dbus.DBUS_INTERFACE, in_signature='s', out_signature=Helper.DBUS_SIGNATURE) @dbus_handle_exceptions def getHelperSettings(self, helper, sender=None): # pylint: disable=W0613 # returns helper settings for helper helper = dbus_to_python(helper, str) log.debug1("getHelperSettings(%s)", helper) return self.fw.helper.get_helper(helper).export_config()	114.52 KB	2026-03-10 23:36
📄 server.py # -- coding: utf-8 -- # # Copyright (C) 2010-2016 Red Hat, Inc. # # Authors: # Thomas Woerner <twoerner@redhat.com> # # This program is free software; you can redistribute it and/or modify # it under the terms of the GNU General Public License as published by # the Free Software Foundation; either version 2 of the License, or # (at your option) any later version. # # This program is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU General Public License for more details. # # You should have received a copy of the GNU General Public License # along with this program. If not, see <http://www.gnu.org/licenses/>. # # signal handling and run_server derived from setroubleshoot # Copyright (C) 2006,2007,2008,2009 Red Hat, Inc. # Authors: # John Dennis <jdennis@redhat.com> # Thomas Liu <tliu@redhat.com> # Dan Walsh <dwalsh@redhat.com> __all__ = [ "run_server" ] import signal from gi.repository import GLib from firewall.core.logger import log from firewall.server.firewalld import FirewallD ############################################################################ # # signal handlers # ############################################################################ def sighup(service): service.reload() return True def sigterm(mainloop): mainloop.quit() ############################################################################ # # run_server function # ############################################################################ def run_server(debug_gc=False): """ Main function for firewall server. Handles D-Bus and GLib mainloop. """ service = None if debug_gc: from pprint import pformat import gc gc.enable() gc.set_debug(gc.DEBUG_LEAK) gc_timeout = 10 def gc_collect(): gc.collect() if len(gc.garbage) > 0: print("\n>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>" ">>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>\n") print("GARBAGE OBJECTS (%d):\n" % len(gc.garbage)) for x in gc.garbage: print(type(x), "\n ",) print(pformat(x)) print("\n<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<" "<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<\n") GLib.timeout_add_seconds(gc_timeout, gc_collect) try: service = FirewallD() mainloop = GLib.MainLoop() if debug_gc: GLib.timeout_add_seconds(gc_timeout, gc_collect) # use unix_signal_add if available, else unix_signal_add_full if hasattr(GLib, 'unix_signal_add'): unix_signal_add = GLib.unix_signal_add else: unix_signal_add = GLib.unix_signal_add_full unix_signal_add(GLib.PRIORITY_HIGH, signal.SIGHUP, sighup, service) unix_signal_add(GLib.PRIORITY_HIGH, signal.SIGTERM, sigterm, mainloop) mainloop.run() except KeyboardInterrupt: log.debug1("Stopping..") except SystemExit: log.error("Raising SystemExit in run_server") except Exception as e: log.error("Exception %s: %s", e.__class__.__name__, str(e)) if service: service.stop()	3.31 KB	2026-03-10 23:36

Terminal