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:56
📄 __init__.py	0 B	2026-03-10 23:36
📁 __pycache__	-	2026-03-21 15:45
📄 client.py # -- coding: utf-8 -- # # Copyright (C) 2009-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/>. # from gi.repository import GLib import dbus.mainloop.glib import functools from firewall import config from firewall.core.base import DEFAULT_ZONE_TARGET, DEFAULT_POLICY_TARGET, DEFAULT_POLICY_PRIORITY from firewall.dbus_utils import dbus_to_python from firewall.core.rich import Rich_Rule from firewall.core.ipset import normalize_ipset_entry, check_entry_overlaps_existing, \ check_for_overlapping_entries from firewall import errors from firewall.errors import FirewallError import dbus import traceback exception_handler = None not_authorized_loop = False def handle_exceptions(func): """Decorator to handle exceptions """ @functools.wraps(func) def _impl(args, kwargs): authorized = False while not authorized: try: return func(args, *kwargs) except dbus.exceptions.DBusException as e: dbus_message = e.get_dbus_message() # returns unicode dbus_name = e.get_dbus_name() if not exception_handler: raise if "NotAuthorizedException" in dbus_name: exception_handler("NotAuthorizedException") elif "org.freedesktop.DBus.Error" in dbus_name: # dbus error, try again exception_handler(dbus_message) else: authorized = True if dbus_message: exception_handler(dbus_message) else: exception_handler(str(e)) except FirewallError as e: if not exception_handler: raise else: exception_handler(str(e)) except Exception: if not exception_handler: raise else: exception_handler(traceback.format_exc()) if not not_authorized_loop: break return _impl # zone config settings class FirewallClientZoneSettings(object): @handle_exceptions def __init__(self, settings = None): self.settings = ["", "", "", False, DEFAULT_ZONE_TARGET, [], [], [], False, [], [], [], [], [], [], False, False] self.settings_name = ["version", "short", "description", "UNUSED", "target", "services", "ports", "icmp_blocks", "masquerade", "forward_ports", "interfaces", "sources", "rules_str", "protocols", "source_ports", "icmp_block_inversion", "forward"] self.settings_dbus_type = ["s", "s", "s", "b", "s", "s", "(ss)", "s", "b", "(ssss)", "s", "s", "s", "s", "(ss)", "b", "b"] if settings: if isinstance(settings, list): for i,v in enumerate(settings): self.settings[i] = settings[i] if isinstance(settings, dict): self.setSettingsDict(settings) @handle_exceptions def __repr__(self): return '%s(%r)' % (self.__class__, self.settings) @handle_exceptions def getSettingsDict(self): settings = {} for key,value in zip(self.settings_name, self.settings): if key == 'UNUSED': continue settings[key] = value return settings @handle_exceptions def setSettingsDict(self, settings): for key in settings: self.settings[self.settings_name.index(key)] = settings[key] @handle_exceptions def getSettingsDbusDict(self): settings = {} for key,value,sig in zip(self.settings_name, self.settings, self.settings_dbus_type): if key == 'UNUSED': continue if type(value) is list: settings[key] = dbus.Array(value, signature=sig) elif type(value) is dict: settings[key] = dbus.Dictionary(value, signature=sig) else: settings[key] = value return settings @handle_exceptions def getRuntimeSettingsDict(self): settings = self.getSettingsDict() # These are not configurable at runtime: del settings['version'] del settings['short'] del settings['description'] del settings['target'] return settings @handle_exceptions def getRuntimeSettingsDbusDict(self): settings = self.getSettingsDbusDict() # These are not configurable at runtime: del settings['version'] del settings['short'] del settings['description'] del settings['target'] return settings @handle_exceptions def getVersion(self): return self.settings[0] @handle_exceptions def setVersion(self, version): self.settings[0] = version @handle_exceptions def getShort(self): return self.settings[1] @handle_exceptions def setShort(self, short): self.settings[1] = short @handle_exceptions def getDescription(self): return self.settings[2] @handle_exceptions def setDescription(self, description): self.settings[2] = description # self.settings[3] was used for 'immutable' @handle_exceptions def getTarget(self): return self.settings[4] if self.settings[4] != DEFAULT_ZONE_TARGET else "default" @handle_exceptions def setTarget(self, target): self.settings[4] = target if target != "default" else DEFAULT_ZONE_TARGET @handle_exceptions def getServices(self): return self.settings[5] @handle_exceptions def setServices(self, services): self.settings[5] = services @handle_exceptions def addService(self, service): if service not in self.settings[5]: self.settings[5].append(service) else: raise FirewallError(errors.ALREADY_ENABLED, service) @handle_exceptions def removeService(self, service): if service in self.settings[5]: self.settings[5].remove(service) else: raise FirewallError(errors.NOT_ENABLED, service) @handle_exceptions def queryService(self, service): return service in self.settings[5] @handle_exceptions def getPorts(self): return self.settings[6] @handle_exceptions def setPorts(self, ports): self.settings[6] = ports @handle_exceptions def addPort(self, port, protocol): if (port,protocol) not in self.settings[6]: self.settings[6].append((port,protocol)) else: raise FirewallError(errors.ALREADY_ENABLED, "'%s:%s'" % (port, protocol)) @handle_exceptions def removePort(self, port, protocol): if (port,protocol) in self.settings[6]: self.settings[6].remove((port,protocol)) else: raise FirewallError(errors.NOT_ENABLED, "'%s:%s'" % (port, protocol)) @handle_exceptions def queryPort(self, port, protocol): return (port,protocol) in self.settings[6] @handle_exceptions def getProtocols(self): return self.settings[13] @handle_exceptions def setProtocols(self, protocols): self.settings[13] = protocols @handle_exceptions def addProtocol(self, protocol): if protocol not in self.settings[13]: self.settings[13].append(protocol) else: raise FirewallError(errors.ALREADY_ENABLED, protocol) @handle_exceptions def removeProtocol(self, protocol): if protocol in self.settings[13]: self.settings[13].remove(protocol) else: raise FirewallError(errors.NOT_ENABLED, protocol) @handle_exceptions def queryProtocol(self, protocol): return protocol in self.settings[13] @handle_exceptions def getSourcePorts(self): return self.settings[14] @handle_exceptions def setSourcePorts(self, ports): self.settings[14] = ports @handle_exceptions def addSourcePort(self, port, protocol): if (port,protocol) not in self.settings[14]: self.settings[14].append((port,protocol)) else: raise FirewallError(errors.ALREADY_ENABLED, "'%s:%s'" % (port, protocol)) @handle_exceptions def removeSourcePort(self, port, protocol): if (port,protocol) in self.settings[14]: self.settings[14].remove((port,protocol)) else: raise FirewallError(errors.NOT_ENABLED, "'%s:%s'" % (port, protocol)) @handle_exceptions def querySourcePort(self, port, protocol): return (port,protocol) in self.settings[14] @handle_exceptions def getIcmpBlocks(self): return self.settings[7] @handle_exceptions def setIcmpBlocks(self, icmpblocks): self.settings[7] = icmpblocks @handle_exceptions def addIcmpBlock(self, icmptype): if icmptype not in self.settings[7]: self.settings[7].append(icmptype) else: raise FirewallError(errors.ALREADY_ENABLED, icmptype) @handle_exceptions def removeIcmpBlock(self, icmptype): if icmptype in self.settings[7]: self.settings[7].remove(icmptype) else: raise FirewallError(errors.NOT_ENABLED, icmptype) @handle_exceptions def queryIcmpBlock(self, icmptype): return icmptype in self.settings[7] @handle_exceptions def getIcmpBlockInversion(self): return self.settings[15] @handle_exceptions def setIcmpBlockInversion(self, flag): self.settings[15] = flag @handle_exceptions def addIcmpBlockInversion(self): if not self.settings[15]: self.settings[15] = True else: raise FirewallError(errors.ALREADY_ENABLED, "icmp-block-inversion") @handle_exceptions def removeIcmpBlockInversion(self): if self.settings[15]: self.settings[15] = False else: raise FirewallError(errors.NOT_ENABLED, "icmp-block-inversion") @handle_exceptions def queryIcmpBlockInversion(self): return self.settings[15] @handle_exceptions def getForward(self): return self.settings[16] @handle_exceptions def setForward(self, forward): self.settings[16] = forward @handle_exceptions def addForward(self): if not self.settings[16]: self.settings[16] = True else: raise FirewallError(errors.ALREADY_ENABLED, "forward") @handle_exceptions def removeForward(self): if self.settings[16]: self.settings[16] = False else: raise FirewallError(errors.NOT_ENABLED, "forward") @handle_exceptions def queryForward(self): return self.settings[16] @handle_exceptions def getMasquerade(self): return self.settings[8] @handle_exceptions def setMasquerade(self, masquerade): self.settings[8] = masquerade @handle_exceptions def addMasquerade(self): if not self.settings[8]: self.settings[8] = True else: raise FirewallError(errors.ALREADY_ENABLED, "masquerade") @handle_exceptions def removeMasquerade(self): if self.settings[8]: self.settings[8] = False else: raise FirewallError(errors.NOT_ENABLED, "masquerade") @handle_exceptions def queryMasquerade(self): return self.settings[8] @handle_exceptions def getForwardPorts(self): return self.settings[9] @handle_exceptions def setForwardPorts(self, ports): self.settings[9] = ports @handle_exceptions def addForwardPort(self, port, protocol, to_port, to_addr): if to_port is None: to_port = '' if to_addr is None: to_addr = '' if (port,protocol,to_port,to_addr) not in self.settings[9]: self.settings[9].append((port,protocol,to_port,to_addr)) else: raise FirewallError(errors.ALREADY_ENABLED, "'%s:%s:%s:%s'" % \ (port, protocol, to_port, to_addr)) @handle_exceptions def removeForwardPort(self, port, protocol, to_port, to_addr): if to_port is None: to_port = '' if to_addr is None: to_addr = '' if (port,protocol,to_port,to_addr) in self.settings[9]: self.settings[9].remove((port,protocol,to_port,to_addr)) else: raise FirewallError(errors.NOT_ENABLED, "'%s:%s:%s:%s'" % \ (port, protocol, to_port, to_addr)) @handle_exceptions def queryForwardPort(self, port, protocol, to_port, to_addr): if to_port is None: to_port = '' if to_addr is None: to_addr = '' return (port,protocol,to_port,to_addr) in self.settings[9] @handle_exceptions def getInterfaces(self): return self.settings[10] @handle_exceptions def setInterfaces(self, interfaces): self.settings[10] = interfaces @handle_exceptions def addInterface(self, interface): if interface not in self.settings[10]: self.settings[10].append(interface) else: raise FirewallError(errors.ALREADY_ENABLED, interface) @handle_exceptions def removeInterface(self, interface): if interface in self.settings[10]: self.settings[10].remove(interface) else: raise FirewallError(errors.NOT_ENABLED, interface) @handle_exceptions def queryInterface(self, interface): return interface in self.settings[10] @handle_exceptions def getSources(self): return self.settings[11] @handle_exceptions def setSources(self, sources): self.settings[11] = sources @handle_exceptions def addSource(self, source): if source not in self.settings[11]: self.settings[11].append(source) else: raise FirewallError(errors.ALREADY_ENABLED, source) @handle_exceptions def removeSource(self, source): if source in self.settings[11]: self.settings[11].remove(source) else: raise FirewallError(errors.NOT_ENABLED, source) @handle_exceptions def querySource(self, source): return source in self.settings[11] @handle_exceptions def getRichRules(self): return self.settings[12] @handle_exceptions def setRichRules(self, rules): rules = [ str(Rich_Rule(rule_str=r)) for r in rules ] self.settings[12] = rules @handle_exceptions def addRichRule(self, rule): rule = str(Rich_Rule(rule_str=rule)) if rule not in self.settings[12]: self.settings[12].append(rule) else: raise FirewallError(errors.ALREADY_ENABLED, rule) @handle_exceptions def removeRichRule(self, rule): rule = str(Rich_Rule(rule_str=rule)) if rule in self.settings[12]: self.settings[12].remove(rule) else: raise FirewallError(errors.NOT_ENABLED, rule) @handle_exceptions def queryRichRule(self, rule): rule = str(Rich_Rule(rule_str=rule)) return rule in self.settings[12] # zone config class FirewallClientConfigZone(object): def __init__(self, bus, path): self.bus = bus self.path = path self.dbus_obj = self.bus.get_object(config.dbus.DBUS_INTERFACE, path) self.fw_zone = dbus.Interface( self.dbus_obj, dbus_interface=config.dbus.DBUS_INTERFACE_CONFIG_ZONE) self.fw_properties = dbus.Interface( self.dbus_obj, dbus_interface='org.freedesktop.DBus.Properties') #TODO: check interface version and revision (need to match client # version) @handle_exceptions def get_property(self, prop): return dbus_to_python(self.fw_properties.Get( config.dbus.DBUS_INTERFACE_CONFIG_ZONE, prop)) @handle_exceptions def get_properties(self): return dbus_to_python(self.fw_properties.GetAll( config.dbus.DBUS_INTERFACE_CONFIG_ZONE)) @handle_exceptions def set_property(self, prop, value): self.fw_properties.Set(config.dbus.DBUS_INTERFACE_CONFIG_ZONE, prop, value) @handle_exceptions def getSettings(self): return FirewallClientZoneSettings(dbus_to_python(self.fw_zone.getSettings2())) @handle_exceptions def update(self, settings): self.fw_zone.update2(settings.getSettingsDbusDict()) @handle_exceptions def loadDefaults(self): self.fw_zone.loadDefaults() @handle_exceptions def remove(self): self.fw_zone.remove() @handle_exceptions def rename(self, name): self.fw_zone.rename(name) # version @handle_exceptions def getVersion(self): return self.fw_zone.getVersion() @handle_exceptions def setVersion(self, version): self.fw_zone.setVersion(version) # short @handle_exceptions def getShort(self): return self.fw_zone.getShort() @handle_exceptions def setShort(self, short): self.fw_zone.setShort(short) # description @handle_exceptions def getDescription(self): return self.fw_zone.getDescription() @handle_exceptions def setDescription(self, description): self.fw_zone.setDescription(description) # target @handle_exceptions def getTarget(self): return self.fw_zone.getTarget() @handle_exceptions def setTarget(self, target): self.fw_zone.setTarget(target) # service @handle_exceptions def getServices(self): return self.fw_zone.getServices() @handle_exceptions def setServices(self, services): self.fw_zone.setServices(services) @handle_exceptions def addService(self, service): self.fw_zone.addService(service) @handle_exceptions def removeService(self, service): self.fw_zone.removeService(service) @handle_exceptions def queryService(self, service): return self.fw_zone.queryService(service) # port @handle_exceptions def getPorts(self): return self.fw_zone.getPorts() @handle_exceptions def setPorts(self, ports): self.fw_zone.setPorts(ports) @handle_exceptions def addPort(self, port, protocol): self.fw_zone.addPort(port, protocol) @handle_exceptions def removePort(self, port, protocol): self.fw_zone.removePort(port, protocol) @handle_exceptions def queryPort(self, port, protocol): return self.fw_zone.queryPort(port, protocol) # protocol @handle_exceptions def getProtocols(self): return self.fw_zone.getProtocols() @handle_exceptions def setProtocols(self, protocols): self.fw_zone.setProtocols(protocols) @handle_exceptions def addProtocol(self, protocol): self.fw_zone.addProtocol(protocol) @handle_exceptions def removeProtocol(self, protocol): self.fw_zone.removeProtocol(protocol) @handle_exceptions def queryProtocol(self, protocol): return self.fw_zone.queryProtocol(protocol) # source-port @handle_exceptions def getSourcePorts(self): return self.fw_zone.getSourcePorts() @handle_exceptions def setSourcePorts(self, ports): self.fw_zone.setSourcePorts(ports) @handle_exceptions def addSourcePort(self, port, protocol): self.fw_zone.addSourcePort(port, protocol) @handle_exceptions def removeSourcePort(self, port, protocol): self.fw_zone.removeSourcePort(port, protocol) @handle_exceptions def querySourcePort(self, port, protocol): return self.fw_zone.querySourcePort(port, protocol) # icmp block @handle_exceptions def getIcmpBlocks(self): return self.fw_zone.getIcmpBlocks() @handle_exceptions def setIcmpBlocks(self, icmptypes): self.fw_zone.setIcmpBlocks(icmptypes) @handle_exceptions def addIcmpBlock(self, icmptype): self.fw_zone.addIcmpBlock(icmptype) @handle_exceptions def removeIcmpBlock(self, icmptype): self.fw_zone.removeIcmpBlock(icmptype) @handle_exceptions def queryIcmpBlock(self, icmptype): return self.fw_zone.queryIcmpBlock(icmptype) # icmp-block-inversion @handle_exceptions def getIcmpBlockInversion(self): return self.fw_zone.getIcmpBlockInversion() @handle_exceptions def setIcmpBlockInversion(self, inversion): self.fw_zone.setIcmpBlockInversion(inversion) @handle_exceptions def addIcmpBlockInversion(self): self.fw_zone.addIcmpBlockInversion() @handle_exceptions def removeIcmpBlockInversion(self): self.fw_zone.removeIcmpBlockInversion() @handle_exceptions def queryIcmpBlockInversion(self): return self.fw_zone.queryIcmpBlockInversion() # forward @handle_exceptions def getForward(self): return self.fw_zone.getSettings2()["forward"] @handle_exceptions def setForward(self, forward): self.fw_zone.update2({"forward": forward}) @handle_exceptions def addForward(self): self.fw_zone.update2({"forward": True}) @handle_exceptions def removeForward(self): self.fw_zone.update2({"forward": False}) @handle_exceptions def queryForward(self): return self.fw_zone.getSettings2()["forward"] # masquerade @handle_exceptions def getMasquerade(self): return self.fw_zone.getMasquerade() @handle_exceptions def setMasquerade(self, masquerade): self.fw_zone.setMasquerade(masquerade) @handle_exceptions def addMasquerade(self): self.fw_zone.addMasquerade() @handle_exceptions def removeMasquerade(self): self.fw_zone.removeMasquerade() @handle_exceptions def queryMasquerade(self): return self.fw_zone.queryMasquerade() # forward port @handle_exceptions def getForwardPorts(self): return self.fw_zone.getForwardPorts() @handle_exceptions def setForwardPorts(self, ports): self.fw_zone.setForwardPorts(ports) @handle_exceptions def addForwardPort(self, port, protocol, toport, toaddr): if toport is None: toport = '' if toaddr is None: toaddr = '' self.fw_zone.addForwardPort(port, protocol, toport, toaddr) @handle_exceptions def removeForwardPort(self, port, protocol, toport, toaddr): if toport is None: toport = '' if toaddr is None: toaddr = '' self.fw_zone.removeForwardPort(port, protocol, toport, toaddr) @handle_exceptions def queryForwardPort(self, port, protocol, toport, toaddr): if toport is None: toport = '' if toaddr is None: toaddr = '' return self.fw_zone.queryForwardPort(port, protocol, toport, toaddr) # interface @handle_exceptions def getInterfaces(self): return self.fw_zone.getInterfaces() @handle_exceptions def setInterfaces(self, interfaces): self.fw_zone.setInterfaces(interfaces) @handle_exceptions def addInterface(self, interface): self.fw_zone.addInterface(interface) @handle_exceptions def removeInterface(self, interface): self.fw_zone.removeInterface(interface) @handle_exceptions def queryInterface(self, interface): return self.fw_zone.queryInterface(interface) # source @handle_exceptions def getSources(self): return self.fw_zone.getSources() @handle_exceptions def setSources(self, sources): self.fw_zone.setSources(sources) @handle_exceptions def addSource(self, source): self.fw_zone.addSource(source) @handle_exceptions def removeSource(self, source): self.fw_zone.removeSource(source) @handle_exceptions def querySource(self, source): return self.fw_zone.querySource(source) # rich rule @handle_exceptions def getRichRules(self): return self.fw_zone.getRichRules() @handle_exceptions def setRichRules(self, rules): self.fw_zone.setRichRules(rules) @handle_exceptions def addRichRule(self, rule): self.fw_zone.addRichRule(rule) @handle_exceptions def removeRichRule(self, rule): self.fw_zone.removeRichRule(rule) @handle_exceptions def queryRichRule(self, rule): return self.fw_zone.queryRichRule(rule) class FirewallClientPolicySettings(object): @handle_exceptions def __init__(self, settings=None): self.settings = {"description": "", "egress_zones": [], "forward_ports": [], "icmp_blocks": [], "ingress_zones": [], "masquerade": False, "ports": [], "priority": DEFAULT_POLICY_PRIORITY, "protocols": [], "rich_rules": [], "services": [], "short": "", "source_ports": [], "target": DEFAULT_POLICY_TARGET, "version": "", } self.settings_dbus_type = ["s", "s", "(ssss)", "s", "s", "b", "(ss)", "i", "s", "s", "s", "s", "(ss)", "s", "s"] if settings: self.setSettingsDict(settings) @handle_exceptions def __repr__(self): return '%s(%r)' % (self.__class__, self.settings) @handle_exceptions def getSettingsDict(self): return self.settings @handle_exceptions def setSettingsDict(self, settings): for key in settings: self.settings[key] = settings[key] @handle_exceptions def getSettingsDbusDict(self): settings = {} for key,sig in zip(self.settings, self.settings_dbus_type): value = self.settings[key] if type(value) is list: settings[key] = dbus.Array(value, signature=sig) elif type(value) is dict: settings[key] = dbus.Dictionary(value, signature=sig) else: settings[key] = value return settings def getRuntimeSettingsDbusDict(self): settings = self.getSettingsDbusDict() for key in ["version", "short", "description", "target"]: del settings[key] return settings @handle_exceptions def getVersion(self): return self.settings["version"] @handle_exceptions def setVersion(self, version): self.settings["version"] = version @handle_exceptions def getShort(self): return self.settings["short"] @handle_exceptions def setShort(self, short): self.settings["short"] = short @handle_exceptions def getDescription(self): return self.settings["description"] @handle_exceptions def setDescription(self, description): self.settings["description"] = description @handle_exceptions def getTarget(self): return self.settings["target"] @handle_exceptions def setTarget(self, target): self.settings["target"] = target @handle_exceptions def getServices(self): return self.settings["services"] @handle_exceptions def setServices(self, services): self.settings["services"] = services @handle_exceptions def addService(self, service): if service not in self.settings["services"]: self.settings["services"].append(service) else: raise FirewallError(errors.ALREADY_ENABLED, service) @handle_exceptions def removeService(self, service): if service in self.settings["services"]: self.settings["services"].remove(service) else: raise FirewallError(errors.NOT_ENABLED, service) @handle_exceptions def queryService(self, service): return service in self.settings["services"] @handle_exceptions def getPorts(self): return self.settings["ports"] @handle_exceptions def setPorts(self, ports): self.settings["ports"] = ports @handle_exceptions def addPort(self, port, protocol): if (port,protocol) not in self.settings["ports"]: self.settings["ports"].append((port,protocol)) else: raise FirewallError(errors.ALREADY_ENABLED, "'%s:%s'" % (port, protocol)) @handle_exceptions def removePort(self, port, protocol): if (port,protocol) in self.settings["ports"]: self.settings["ports"].remove((port,protocol)) else: raise FirewallError(errors.NOT_ENABLED, "'%s:%s'" % (port, protocol)) @handle_exceptions def queryPort(self, port, protocol): return (port,protocol) in self.settings["ports"] @handle_exceptions def getProtocols(self): return self.settings["protocols"] @handle_exceptions def setProtocols(self, protocols): self.settings["protocols"] = protocols @handle_exceptions def addProtocol(self, protocol): if protocol not in self.settings["protocols"]: self.settings["protocols"].append(protocol) else: raise FirewallError(errors.ALREADY_ENABLED, protocol) @handle_exceptions def removeProtocol(self, protocol): if protocol in self.settings["protocols"]: self.settings["protocols"].remove(protocol) else: raise FirewallError(errors.NOT_ENABLED, protocol) @handle_exceptions def queryProtocol(self, protocol): return protocol in self.settings["protocols"] @handle_exceptions def getSourcePorts(self): return self.settings["source_ports"] @handle_exceptions def setSourcePorts(self, ports): self.settings["source_ports"] = ports @handle_exceptions def addSourcePort(self, port, protocol): if (port,protocol) not in self.settings["source_ports"]: self.settings["source_ports"].append((port,protocol)) else: raise FirewallError(errors.ALREADY_ENABLED, "'%s:%s'" % (port, protocol)) @handle_exceptions def removeSourcePort(self, port, protocol): if (port,protocol) in self.settings["source_ports"]: self.settings["source_ports"].remove((port,protocol)) else: raise FirewallError(errors.NOT_ENABLED, "'%s:%s'" % (port, protocol)) @handle_exceptions def querySourcePort(self, port, protocol): return (port,protocol) in self.settings["source_ports"] @handle_exceptions def getIcmpBlocks(self): return self.settings["icmp_blocks"] @handle_exceptions def setIcmpBlocks(self, icmpblocks): self.settings["icmp_blocks"] = icmpblocks @handle_exceptions def addIcmpBlock(self, icmptype): if icmptype not in self.settings["icmp_blocks"]: self.settings["icmp_blocks"].append(icmptype) else: raise FirewallError(errors.ALREADY_ENABLED, icmptype) @handle_exceptions def removeIcmpBlock(self, icmptype): if icmptype in self.settings["icmp_blocks"]: self.settings["icmp_blocks"].remove(icmptype) else: raise FirewallError(errors.NOT_ENABLED, icmptype) @handle_exceptions def queryIcmpBlock(self, icmptype): return icmptype in self.settings["icmp_blocks"] @handle_exceptions def getMasquerade(self): return self.settings["masquerade"] @handle_exceptions def setMasquerade(self, masquerade): self.settings["masquerade"] = masquerade @handle_exceptions def addMasquerade(self): if not self.settings["masquerade"]: self.settings["masquerade"] = True else: raise FirewallError(errors.ALREADY_ENABLED, "masquerade") @handle_exceptions def removeMasquerade(self): if self.settings["masquerade"]: self.settings["masquerade"] = False else: raise FirewallError(errors.NOT_ENABLED, "masquerade") @handle_exceptions def queryMasquerade(self): return self.settings["masquerade"] @handle_exceptions def getForwardPorts(self): return self.settings["forward_ports"] @handle_exceptions def setForwardPorts(self, ports): self.settings["forward_ports"] = ports @handle_exceptions def addForwardPort(self, port, protocol, to_port, to_addr): if to_port is None: to_port = '' if to_addr is None: to_addr = '' if (port,protocol,to_port,to_addr) not in self.settings["forward_ports"]: self.settings["forward_ports"].append((port,protocol,to_port,to_addr)) else: raise FirewallError(errors.ALREADY_ENABLED, "'%s:%s:%s:%s'" % \ (port, protocol, to_port, to_addr)) @handle_exceptions def removeForwardPort(self, port, protocol, to_port, to_addr): if to_port is None: to_port = '' if to_addr is None: to_addr = '' if (port,protocol,to_port,to_addr) in self.settings["forward_ports"]: self.settings["forward_ports"].remove((port,protocol,to_port,to_addr)) else: raise FirewallError(errors.NOT_ENABLED, "'%s:%s:%s:%s'" % \ (port, protocol, to_port, to_addr)) @handle_exceptions def queryForwardPort(self, port, protocol, to_port, to_addr): if to_port is None: to_port = '' if to_addr is None: to_addr = '' return (port,protocol,to_port,to_addr) in self.settings["forward_ports"] @handle_exceptions def getRichRules(self): return self.settings["rich_rules"] @handle_exceptions def setRichRules(self, rules): rules = [ str(Rich_Rule(rule_str=r)) for r in rules ] self.settings["rich_rules"] = rules @handle_exceptions def addRichRule(self, rule): rule = str(Rich_Rule(rule_str=rule)) if rule not in self.settings["rich_rules"]: self.settings["rich_rules"].append(rule) else: raise FirewallError(errors.ALREADY_ENABLED, rule) @handle_exceptions def removeRichRule(self, rule): rule = str(Rich_Rule(rule_str=rule)) if rule in self.settings["rich_rules"]: self.settings["rich_rules"].remove(rule) else: raise FirewallError(errors.NOT_ENABLED, rule) @handle_exceptions def queryRichRule(self, rule): rule = str(Rich_Rule(rule_str=rule)) return rule in self.settings["rich_rules"] @handle_exceptions def getIngressZones(self): return self.settings["ingress_zones"] @handle_exceptions def setIngressZones(self, ingress_zones): self.settings["ingress_zones"] = ingress_zones @handle_exceptions def addIngressZone(self, ingress_zone): if ingress_zone not in self.settings["ingress_zones"]: self.settings["ingress_zones"].append(ingress_zone) else: raise FirewallError(errors.ALREADY_ENABLED, ingress_zone) @handle_exceptions def removeIngressZone(self, ingress_zone): if ingress_zone in self.settings["ingress_zones"]: self.settings["ingress_zones"].remove(ingress_zone) else: raise FirewallError(errors.NOT_ENABLED, ingress_zone) @handle_exceptions def queryIngressZone(self, ingress_zone): return ingress_zone in self.settings["ingress_zones"] @handle_exceptions def getEgressZones(self): return self.settings["egress_zones"] @handle_exceptions def setEgressZones(self, egress_zones): self.settings["egress_zones"] = egress_zones @handle_exceptions def addEgressZone(self, egress_zone): if egress_zone not in self.settings["egress_zones"]: self.settings["egress_zones"].append(egress_zone) else: raise FirewallError(errors.ALREADY_ENABLED, egress_zone) @handle_exceptions def removeEgressZone(self, egress_zone): if egress_zone in self.settings["egress_zones"]: self.settings["egress_zones"].remove(egress_zone) else: raise FirewallError(errors.NOT_ENABLED, egress_zone) @handle_exceptions def queryEgressZone(self, egress_zone): return egress_zone in self.settings["egress_zones"] @handle_exceptions def getPriority(self): return self.settings["priority"] @handle_exceptions def setPriority(self, priority): self.settings["priority"] = int(priority) class FirewallClientConfigPolicy(object): def __init__(self, bus, path): self.bus = bus self.path = path self.dbus_obj = self.bus.get_object(config.dbus.DBUS_INTERFACE, path) self.fw_policy = dbus.Interface( self.dbus_obj, dbus_interface=config.dbus.DBUS_INTERFACE_CONFIG_POLICY) self.fw_properties = dbus.Interface( self.dbus_obj, dbus_interface='org.freedesktop.DBus.Properties') @handle_exceptions def get_property(self, prop): return dbus_to_python(self.fw_properties.Get( config.dbus.DBUS_INTERFACE_CONFIG_POLICY, prop)) @handle_exceptions def get_properties(self): return dbus_to_python(self.fw_properties.GetAll( config.dbus.DBUS_INTERFACE_CONFIG_POLICY)) @handle_exceptions def set_property(self, prop, value): self.fw_properties.Set(config.dbus.DBUS_INTERFACE_CONFIG_POLICY, prop, value) @handle_exceptions def getSettings(self): return FirewallClientPolicySettings(dbus_to_python(self.fw_policy.getSettings())) @handle_exceptions def update(self, settings): self.fw_policy.update(settings.getSettingsDbusDict()) @handle_exceptions def loadDefaults(self): self.fw_policy.loadDefaults() @handle_exceptions def remove(self): self.fw_policy.remove() @handle_exceptions def rename(self, name): self.fw_policy.rename(name) # service config settings class FirewallClientServiceSettings(object): @handle_exceptions def __init__(self, settings=None): self.settings = ["", "", "", [], [], {}, [], [], [], []] self.settings_name = ["version", "short", "description", "ports", "modules", "destination", "protocols", "source_ports", "includes", "helpers"] self.settings_dbus_type = ["s", "s", "s", "(ss)", "s", "ss", "s", "(ss)", "s", "s"] if settings: if type(settings) is list: for i,v in enumerate(settings): self.settings[i] = settings[i] elif type(settings) is dict: self.setSettingsDict(settings) @handle_exceptions def __repr__(self): return '%s(%r)' % (self.__class__, self.settings) @handle_exceptions def getSettingsDict(self): settings = {} for key,value in zip(self.settings_name, self.settings): settings[key] = value return settings @handle_exceptions def setSettingsDict(self, settings): for key in settings: self.settings[self.settings_name.index(key)] = settings[key] @handle_exceptions def getSettingsDbusDict(self): settings = {} for key,value,sig in zip(self.settings_name, self.settings, self.settings_dbus_type): if type(value) is list: settings[key] = dbus.Array(value, signature=sig) elif type(value) is dict: settings[key] = dbus.Dictionary(value, signature=sig) else: settings[key] = value return settings @handle_exceptions def getVersion(self): return self.settings[0] @handle_exceptions def setVersion(self, version): self.settings[0] = version @handle_exceptions def getShort(self): return self.settings[1] @handle_exceptions def setShort(self, short): self.settings[1] = short @handle_exceptions def getDescription(self): return self.settings[2] @handle_exceptions def setDescription(self, description): self.settings[2] = description @handle_exceptions def getPorts(self): return self.settings[3] @handle_exceptions def setPorts(self, ports): self.settings[3] = ports @handle_exceptions def addPort(self, port, protocol): if (port,protocol) not in self.settings[3]: self.settings[3].append((port,protocol)) else: raise FirewallError(errors.ALREADY_ENABLED, "'%s:%s'" % (port, protocol)) @handle_exceptions def removePort(self, port, protocol): if (port,protocol) in self.settings[3]: self.settings[3].remove((port,protocol)) else: raise FirewallError(errors.NOT_ENABLED, "'%s:%s'" % (port, protocol)) @handle_exceptions def queryPort(self, port, protocol): return (port,protocol) in self.settings[3] @handle_exceptions def getProtocols(self): return self.settings[6] @handle_exceptions def setProtocols(self, protocols): self.settings[6] = protocols @handle_exceptions def addProtocol(self, protocol): if protocol not in self.settings[6]: self.settings[6].append(protocol) else: raise FirewallError(errors.ALREADY_ENABLED, protocol) @handle_exceptions def removeProtocol(self, protocol): if protocol in self.settings[6]: self.settings[6].remove(protocol) else: raise FirewallError(errors.NOT_ENABLED, protocol) @handle_exceptions def queryProtocol(self, protocol): return protocol in self.settings[6] @handle_exceptions def getSourcePorts(self): return self.settings[7] @handle_exceptions def setSourcePorts(self, ports): self.settings[7] = ports @handle_exceptions def addSourcePort(self, port, protocol): if (port,protocol) not in self.settings[7]: self.settings[7].append((port,protocol)) else: raise FirewallError(errors.ALREADY_ENABLED, "'%s:%s'" % (port, protocol)) @handle_exceptions def removeSourcePort(self, port, protocol): if (port,protocol) in self.settings[7]: self.settings[7].remove((port,protocol)) else: raise FirewallError(errors.NOT_ENABLED, "'%s:%s'" % (port, protocol)) @handle_exceptions def querySourcePort(self, port, protocol): return (port,protocol) in self.settings[7] @handle_exceptions def getModules(self): return self.settings[4] @handle_exceptions def setModules(self, modules): self.settings[4] = modules @handle_exceptions def addModule(self, module): if module not in self.settings[4]: self.settings[4].append(module) else: raise FirewallError(errors.ALREADY_ENABLED, module) @handle_exceptions def removeModule(self, module): if module in self.settings[4]: self.settings[4].remove(module) else: raise FirewallError(errors.NOT_ENABLED, module) @handle_exceptions def queryModule(self, module): return module in self.settings[4] @handle_exceptions def getDestinations(self): return self.settings[5] @handle_exceptions def setDestinations(self, destinations): self.settings[5] = destinations @handle_exceptions def setDestination(self, dest_type, address): if dest_type not in self.settings[5] or \ self.settings[5][dest_type] != address: self.settings[5][dest_type] = address else: raise FirewallError(errors.ALREADY_ENABLED, "'%s:%s'" % \ (dest_type, address)) @handle_exceptions def removeDestination(self, dest_type, address=None): if dest_type in self.settings[5]: if address is not None and self.settings[5][dest_type] != address: raise FirewallError(errors.NOT_ENABLED, "'%s:%s'" % \ (dest_type, address)) del self.settings[5][dest_type] else: raise FirewallError(errors.NOT_ENABLED, "'%s'" % dest_type) @handle_exceptions def queryDestination(self, dest_type, address): return (dest_type in self.settings[5] and \ address == self.settings[5][dest_type]) @handle_exceptions def getIncludes(self): return self.settings[8] @handle_exceptions def setIncludes(self, includes): self.settings[8] = includes @handle_exceptions def addInclude(self, include): if include not in self.settings[8]: self.settings[8].append(include) else: raise FirewallError(errors.ALREADY_ENABLED, include) @handle_exceptions def removeInclude(self, include): if include in self.settings[8]: self.settings[8].remove(include) else: raise FirewallError(errors.NOT_ENABLED, include) @handle_exceptions def queryInclude(self, include): return include in self.settings[8] @handle_exceptions def getHelpers(self): return self.settings[9] @handle_exceptions def setHelpers(self, helpers): self.settings[9] = helpers @handle_exceptions def addHelper(self, helper): if helper not in self.settings[9]: self.settings[9].append(helper) else: raise FirewallError(errors.ALREADY_ENABLED, helper) @handle_exceptions def removeHelper(self, helper): if helper in self.settings[9]: self.settings[9].remove(helper) else: raise FirewallError(errors.NOT_ENABLED, helper) @handle_exceptions def queryHelper(self, helper): return helper in self.settings[9] # ipset config settings class FirewallClientIPSetSettings(object): @handle_exceptions def __init__(self, settings=None): if settings: self.settings = settings else: self.settings = ["", "", "", "", {}, []] @handle_exceptions def __repr__(self): return '%s(%r)' % (self.__class__, self.settings) @handle_exceptions def getVersion(self): return self.settings[0] @handle_exceptions def setVersion(self, version): self.settings[0] = version @handle_exceptions def getShort(self): return self.settings[1] @handle_exceptions def setShort(self, short): self.settings[1] = short @handle_exceptions def getDescription(self): return self.settings[2] @handle_exceptions def setDescription(self, description): self.settings[2] = description @handle_exceptions def getType(self): return self.settings[3] @handle_exceptions def setType(self, ipset_type): self.settings[3] = ipset_type @handle_exceptions def getOptions(self): return self.settings[4] @handle_exceptions def setOptions(self, options): self.settings[4] = options @handle_exceptions def addOption(self, key, value): if key not in self.settings[4] or self.settings[4][key] != value: self.settings[4][key] = value else: raise FirewallError(errors.ALREADY_ENABLED, "'%s=%s'" % (key,value) if value else key) @handle_exceptions def removeOption(self, key): if key in self.settings[4]: del self.settings[4][key] else: raise FirewallError(errors.NOT_ENABLED, key) @handle_exceptions def queryOption(self, key, value): return key in self.settings[4] and self.settings[4][key] == value @handle_exceptions def getEntries(self): return self.settings[5] @handle_exceptions def setEntries(self, entries): if "timeout" in self.settings[4] and \ self.settings[4]["timeout"] != "0": raise FirewallError(errors.IPSET_WITH_TIMEOUT) check_for_overlapping_entries(entries) self.settings[5] = entries @handle_exceptions def addEntry(self, entry): if "timeout" in self.settings[4] and \ self.settings[4]["timeout"] != "0": raise FirewallError(errors.IPSET_WITH_TIMEOUT) entry = normalize_ipset_entry(entry) if entry not in self.settings[5]: check_entry_overlaps_existing(entry, self.settings[5]) self.settings[5].append(entry) else: raise FirewallError(errors.ALREADY_ENABLED, entry) @handle_exceptions def removeEntry(self, entry): if "timeout" in self.settings[4] and \ self.settings[4]["timeout"] != "0": raise FirewallError(errors.IPSET_WITH_TIMEOUT) entry = normalize_ipset_entry(entry) if entry in self.settings[5]: self.settings[5].remove(entry) else: raise FirewallError(errors.NOT_ENABLED, entry) @handle_exceptions def queryEntry(self, entry): if "timeout" in self.settings[4] and \ self.settings[4]["timeout"] != "0": raise FirewallError(errors.IPSET_WITH_TIMEOUT) entry = normalize_ipset_entry(entry) return entry in self.settings[5] # ipset config class FirewallClientConfigIPSet(object): @handle_exceptions def __init__(self, bus, path): self.bus = bus self.path = path self.dbus_obj = self.bus.get_object(config.dbus.DBUS_INTERFACE, path) self.fw_ipset = dbus.Interface( self.dbus_obj, dbus_interface=config.dbus.DBUS_INTERFACE_CONFIG_IPSET) self.fw_properties = dbus.Interface( self.dbus_obj, dbus_interface='org.freedesktop.DBus.Properties') @handle_exceptions def get_property(self, prop): return dbus_to_python(self.fw_properties.Get( config.dbus.DBUS_INTERFACE_CONFIG_IPSET, prop)) @handle_exceptions def get_properties(self): return dbus_to_python(self.fw_properties.GetAll( config.dbus.DBUS_INTERFACE_CONFIG_IPSET)) @handle_exceptions def set_property(self, prop, value): self.fw_properties.Set(config.dbus.DBUS_INTERFACE_CONFIG_IPSET, prop, value) @handle_exceptions def getSettings(self): return FirewallClientIPSetSettings(list(dbus_to_python(\ self.fw_ipset.getSettings()))) @handle_exceptions def update(self, settings): self.fw_ipset.update(tuple(settings.settings)) @handle_exceptions def loadDefaults(self): self.fw_ipset.loadDefaults() @handle_exceptions def remove(self): self.fw_ipset.remove() @handle_exceptions def rename(self, name): self.fw_ipset.rename(name) # version @handle_exceptions def getVersion(self): return self.fw_ipset.getVersion() @handle_exceptions def setVersion(self, version): self.fw_ipset.setVersion(version) # short @handle_exceptions def getShort(self): return self.fw_ipset.getShort() @handle_exceptions def setShort(self, short): self.fw_ipset.setShort(short) # description @handle_exceptions def getDescription(self): return self.fw_ipset.getDescription() @handle_exceptions def setDescription(self, description): self.fw_ipset.setDescription(description) # entry @handle_exceptions def getEntries(self): return self.fw_ipset.getEntries() @handle_exceptions def setEntries(self, entries): self.fw_ipset.setEntries(entries) @handle_exceptions def addEntry(self, entry): self.fw_ipset.addEntry(entry) @handle_exceptions def removeEntry(self, entry): self.fw_ipset.removeEntry(entry) @handle_exceptions def queryEntry(self, entry): return self.fw_ipset.queryEntry(entry) # helper config settings class FirewallClientHelperSettings(object): @handle_exceptions def __init__(self, settings=None): if settings: self.settings = settings else: self.settings = ["", "", "", "", "", [ ]] @handle_exceptions def __repr__(self): return '%s(%r)' % (self.__class__, self.settings) @handle_exceptions def getVersion(self): return self.settings[0] @handle_exceptions def setVersion(self, version): self.settings[0] = version @handle_exceptions def getShort(self): return self.settings[1] @handle_exceptions def setShort(self, short): self.settings[1] = short @handle_exceptions def getDescription(self): return self.settings[2] @handle_exceptions def setDescription(self, description): self.settings[2] = description @handle_exceptions def getFamily(self): return self.settings[3] @handle_exceptions def setFamily(self, ipv): if ipv is None: self.settings[3] = "" self.settings[3] = ipv @handle_exceptions def getModule(self): return self.settings[4] @handle_exceptions def setModule(self, module): self.settings[4] = module @handle_exceptions def getPorts(self): return self.settings[5] @handle_exceptions def setPorts(self, ports): self.settings[5] = ports @handle_exceptions def addPort(self, port, protocol): if (port,protocol) not in self.settings[5]: self.settings[5].append((port,protocol)) else: raise FirewallError(errors.ALREADY_ENABLED, "'%s:%s'" % (port, protocol)) @handle_exceptions def removePort(self, port, protocol): if (port,protocol) in self.settings[5]: self.settings[5].remove((port,protocol)) else: raise FirewallError(errors.NOT_ENABLED, "'%s:%s'" % (port, protocol)) @handle_exceptions def queryPort(self, port, protocol): return (port,protocol) in self.settings[5] # helper config class FirewallClientConfigHelper(object): @handle_exceptions def __init__(self, bus, path): self.bus = bus self.path = path self.dbus_obj = self.bus.get_object(config.dbus.DBUS_INTERFACE, path) self.fw_helper = dbus.Interface( self.dbus_obj, dbus_interface=config.dbus.DBUS_INTERFACE_CONFIG_HELPER) self.fw_properties = dbus.Interface( self.dbus_obj, dbus_interface='org.freedesktop.DBus.Properties') @handle_exceptions def get_property(self, prop): return dbus_to_python(self.fw_properties.Get( config.dbus.DBUS_INTERFACE_CONFIG_HELPER, prop)) @handle_exceptions def get_properties(self): return dbus_to_python(self.fw_properties.GetAll( config.dbus.DBUS_INTERFACE_CONFIG_HELPER)) @handle_exceptions def set_property(self, prop, value): self.fw_properties.Set(config.dbus.DBUS_INTERFACE_CONFIG_HELPER, prop, value) @handle_exceptions def getSettings(self): return FirewallClientHelperSettings(list(dbus_to_python(\ self.fw_helper.getSettings()))) @handle_exceptions def update(self, settings): self.fw_helper.update(tuple(settings.settings)) @handle_exceptions def loadDefaults(self): self.fw_helper.loadDefaults() @handle_exceptions def remove(self): self.fw_helper.remove() @handle_exceptions def rename(self, name): self.fw_helper.rename(name) # version @handle_exceptions def getVersion(self): return self.fw_helper.getVersion() @handle_exceptions def setVersion(self, version): self.fw_helper.setVersion(version) # short @handle_exceptions def getShort(self): return self.fw_helper.getShort() @handle_exceptions def setShort(self, short): self.fw_helper.setShort(short) # description @handle_exceptions def getDescription(self): return self.fw_helper.getDescription() @handle_exceptions def setDescription(self, description): self.fw_helper.setDescription(description) # port @handle_exceptions def getPorts(self): return self.fw_helper.getPorts() @handle_exceptions def setPorts(self, ports): self.fw_helper.setPorts(ports) @handle_exceptions def addPort(self, port, protocol): self.fw_helper.addPort(port, protocol) @handle_exceptions def removePort(self, port, protocol): self.fw_helper.removePort(port, protocol) @handle_exceptions def queryPort(self, port, protocol): return self.fw_helper.queryPort(port, protocol) # family @handle_exceptions def getFamily(self): return self.fw_helper.getFamily() @handle_exceptions def setFamily(self, ipv): if ipv is None: self.fw_helper.setFamily("") self.fw_helper.setFamily(ipv) # module @handle_exceptions def getModule(self): return self.fw_helper.getModule() @handle_exceptions def setModule(self, module): self.fw_helper.setModule(module) # service config class FirewallClientConfigService(object): @handle_exceptions def __init__(self, bus, path): self.bus = bus self.path = path self.dbus_obj = self.bus.get_object(config.dbus.DBUS_INTERFACE, path) self.fw_service = dbus.Interface( self.dbus_obj, dbus_interface=config.dbus.DBUS_INTERFACE_CONFIG_SERVICE) self.fw_properties = dbus.Interface( self.dbus_obj, dbus_interface='org.freedesktop.DBus.Properties') @handle_exceptions def get_property(self, prop): return dbus_to_python(self.fw_properties.Get( config.dbus.DBUS_INTERFACE_CONFIG_SERVICE, prop)) @handle_exceptions def get_properties(self): return dbus_to_python(self.fw_properties.GetAll( config.dbus.DBUS_INTERFACE_CONFIG_SERVICE)) @handle_exceptions def set_property(self, prop, value): self.fw_properties.Set(config.dbus.DBUS_INTERFACE_CONFIG_SERVICE, prop, value) @handle_exceptions def getSettings(self): return FirewallClientServiceSettings(dbus_to_python( self.fw_service.getSettings2())) @handle_exceptions def update(self, settings): self.fw_service.update2(settings.getSettingsDbusDict()) @handle_exceptions def loadDefaults(self): self.fw_service.loadDefaults() @handle_exceptions def remove(self): self.fw_service.remove() @handle_exceptions def rename(self, name): self.fw_service.rename(name) # version @handle_exceptions def getVersion(self): return self.fw_service.getVersion() @handle_exceptions def setVersion(self, version): self.fw_service.setVersion(version) # short @handle_exceptions def getShort(self): return self.fw_service.getShort() @handle_exceptions def setShort(self, short): self.fw_service.setShort(short) # description @handle_exceptions def getDescription(self): return self.fw_service.getDescription() @handle_exceptions def setDescription(self, description): self.fw_service.setDescription(description) # port @handle_exceptions def getPorts(self): return self.fw_service.getPorts() @handle_exceptions def setPorts(self, ports): self.fw_service.setPorts(ports) @handle_exceptions def addPort(self, port, protocol): self.fw_service.addPort(port, protocol) @handle_exceptions def removePort(self, port, protocol): self.fw_service.removePort(port, protocol) @handle_exceptions def queryPort(self, port, protocol): return self.fw_service.queryPort(port, protocol) # protocol @handle_exceptions def getProtocols(self): return self.fw_service.getProtocols() @handle_exceptions def setProtocols(self, protocols): self.fw_service.setProtocols(protocols) @handle_exceptions def addProtocol(self, protocol): self.fw_service.addProtocol(protocol) @handle_exceptions def removeProtocol(self, protocol): self.fw_service.removeProtocol(protocol) @handle_exceptions def queryProtocol(self, protocol): return self.fw_service.queryProtocol(protocol) # source-port @handle_exceptions def getSourcePorts(self): return self.fw_service.getSourcePorts() @handle_exceptions def setSourcePorts(self, ports): self.fw_service.setSourcePorts(ports) @handle_exceptions def addSourcePort(self, port, protocol): self.fw_service.addSourcePort(port, protocol) @handle_exceptions def removeSourcePort(self, port, protocol): self.fw_service.removeSourcePort(port, protocol) @handle_exceptions def querySourcePort(self, port, protocol): return self.fw_service.querySourcePort(port, protocol) # module @handle_exceptions def getModules(self): return self.fw_service.getModules() @handle_exceptions def setModules(self, modules): self.fw_service.setModules(modules) @handle_exceptions def addModule(self, module): self.fw_service.addModule(module) @handle_exceptions def removeModule(self, module): self.fw_service.removeModule(module) @handle_exceptions def queryModule(self, module): return self.fw_service.queryModule(module) # destination @handle_exceptions def getDestinations(self): return self.fw_service.getDestinations() @handle_exceptions def setDestinations(self, destinations): self.fw_service.setDestinations(destinations) @handle_exceptions def getDestination(self, destination): return self.fw_service.getDestination(destination) @handle_exceptions def setDestination(self, destination, address): self.fw_service.setDestination(destination, address) @handle_exceptions def removeDestination(self, destination, address=None): if address is not None and self.getDestination(destination) != address: raise FirewallError(errors.NOT_ENABLED, "'%s:%s'" % \ (destination, address)) self.fw_service.removeDestination(destination) @handle_exceptions def queryDestination(self, destination, address): return self.fw_service.queryDestination(destination, address) # include @handle_exceptions def getIncludes(self): return self.fw_service.getIncludes() @handle_exceptions def setIncludes(self, includes): self.fw_service.setIncludes(includes) @handle_exceptions def addInclude(self, include): self.fw_service.addInclude(include) @handle_exceptions def removeInclude(self, include): self.fw_service.removeInclude(include) @handle_exceptions def queryInclude(self, include): return self.fw_service.queryInclude(include) # icmptype config settings class FirewallClientIcmpTypeSettings(object): @handle_exceptions def __init__(self, settings=None): if settings: self.settings = settings else: self.settings = ["", "", "", []] @handle_exceptions def __repr__(self): return '%s(%r)' % (self.__class__, self.settings) @handle_exceptions def getVersion(self): return self.settings[0] @handle_exceptions def setVersion(self, version): self.settings[0] = version @handle_exceptions def getShort(self): return self.settings[1] @handle_exceptions def setShort(self, short): self.settings[1] = short @handle_exceptions def getDescription(self): return self.settings[2] @handle_exceptions def setDescription(self, description): self.settings[2] = description @handle_exceptions def getDestinations(self): return self.settings[3] @handle_exceptions def setDestinations(self, destinations): self.settings[3] = destinations @handle_exceptions def addDestination(self, destination): # empty means all if not self.settings[3]: raise FirewallError(errors.ALREADY_ENABLED, destination) elif destination not in self.settings[3]: self.settings[3].append(destination) else: raise FirewallError(errors.ALREADY_ENABLED, destination) @handle_exceptions def removeDestination(self, destination): if destination in self.settings[3]: self.settings[3].remove(destination) # empty means all elif not self.settings[3]: self.setDestinations(list(set(['ipv4','ipv6']) - \ set([destination]))) else: raise FirewallError(errors.NOT_ENABLED, destination) @handle_exceptions def queryDestination(self, destination): # empty means all return not self.settings[3] or \ destination in self.settings[3] # icmptype config class FirewallClientConfigIcmpType(object): @handle_exceptions def __init__(self, bus, path): self.bus = bus self.path = path self.dbus_obj = self.bus.get_object(config.dbus.DBUS_INTERFACE, path) self.fw_icmptype = dbus.Interface( self.dbus_obj, dbus_interface=config.dbus.DBUS_INTERFACE_CONFIG_ICMPTYPE) self.fw_properties = dbus.Interface( self.dbus_obj, dbus_interface='org.freedesktop.DBus.Properties') @handle_exceptions def get_property(self, prop): return dbus_to_python(self.fw_properties.Get( config.dbus.DBUS_INTERFACE_CONFIG_ICMPTYPE, prop)) @handle_exceptions def get_properties(self): return dbus_to_python(self.fw_properties.GetAll( config.dbus.DBUS_INTERFACE_CONFIG_ICMPTYPE)) @handle_exceptions def set_property(self, prop, value): self.fw_properties.Set(config.dbus.DBUS_INTERFACE_CONFIG_ICMPTYPE, prop, value) @handle_exceptions def getSettings(self): return FirewallClientIcmpTypeSettings(list(dbus_to_python(\ self.fw_icmptype.getSettings()))) @handle_exceptions def update(self, settings): self.fw_icmptype.update(tuple(settings.settings)) @handle_exceptions def loadDefaults(self): self.fw_icmptype.loadDefaults() @handle_exceptions def remove(self): self.fw_icmptype.remove() @handle_exceptions def rename(self, name): self.fw_icmptype.rename(name) # version @handle_exceptions def getVersion(self): return self.fw_icmptype.getVersion() @handle_exceptions def setVersion(self, version): self.fw_icmptype.setVersion(version) # short @handle_exceptions def getShort(self): return self.fw_icmptype.getShort() @handle_exceptions def setShort(self, short): self.fw_icmptype.setShort(short) # description @handle_exceptions def getDescription(self): return self.fw_icmptype.getDescription() @handle_exceptions def setDescription(self, description): self.fw_icmptype.setDescription(description) # destination @handle_exceptions def getDestinations(self): return self.fw_icmptype.getDestinations() @handle_exceptions def setDestinations(self, destinations): self.fw_icmptype.setDestinations(destinations) @handle_exceptions def addDestination(self, destination): self.fw_icmptype.addDestination(destination) @handle_exceptions def removeDestination(self, destination): self.fw_icmptype.removeDestination(destination) @handle_exceptions def queryDestination(self, destination): return self.fw_icmptype.queryDestination(destination) # config.policies lockdown whitelist class FirewallClientPoliciesLockdownWhitelist(object): @handle_exceptions def __init__(self, settings=None): if settings: self.settings = settings else: self.settings = [ [], [], [], [] ] @handle_exceptions def __repr__(self): return '%s(%r)' % (self.__class__, self.settings) @handle_exceptions def getCommands(self): return self.settings[0] @handle_exceptions def setCommands(self, commands): self.settings[0] = commands @handle_exceptions def addCommand(self, command): if command not in self.settings[0]: self.settings[0].append(command) @handle_exceptions def removeCommand(self, command): if command in self.settings[0]: self.settings[0].remove(command) @handle_exceptions def queryCommand(self, command): return command in self.settings[0] @handle_exceptions def getContexts(self): return self.settings[1] @handle_exceptions def setContexts(self, contexts): self.settings[1] = contexts @handle_exceptions def addContext(self, context): if context not in self.settings[1]: self.settings[1].append(context) @handle_exceptions def removeContext(self, context): if context in self.settings[1]: self.settings[1].remove(context) @handle_exceptions def queryContext(self, context): return context in self.settings[1] @handle_exceptions def getUsers(self): return self.settings[2] @handle_exceptions def setUsers(self, users): self.settings[2] = users @handle_exceptions def addUser(self, user): if user not in self.settings[2]: self.settings[2].append(user) @handle_exceptions def removeUser(self, user): if user in self.settings[2]: self.settings[2].remove(user) @handle_exceptions def queryUser(self, user): return user in self.settings[2] @handle_exceptions def getUids(self): return self.settings[3] @handle_exceptions def setUids(self, uids): self.settings[3] = uids @handle_exceptions def addUid(self, uid): if uid not in self.settings[3]: self.settings[3].append(uid) @handle_exceptions def removeUid(self, uid): if uid in self.settings[3]: self.settings[3].remove(uid) @handle_exceptions def queryUid(self, uid): return uid in self.settings[3] # config.policies class FirewallClientConfigPolicies(object): @handle_exceptions def __init__(self, bus): self.bus = bus self.dbus_obj = self.bus.get_object(config.dbus.DBUS_INTERFACE, config.dbus.DBUS_PATH_CONFIG) self.fw_policies = dbus.Interface( self.dbus_obj, dbus_interface=config.dbus.DBUS_INTERFACE_CONFIG_POLICIES) @handle_exceptions def getLockdownWhitelist(self): return FirewallClientPoliciesLockdownWhitelist( \ list(dbus_to_python(self.fw_policies.getLockdownWhitelist()))) @handle_exceptions def setLockdownWhitelist(self, settings): self.fw_policies.setLockdownWhitelist(tuple(settings.settings)) # command @handle_exceptions def addLockdownWhitelistCommand(self, command): self.fw_policies.addLockdownWhitelistCommand(command) @handle_exceptions def removeLockdownWhitelistCommand(self, command): self.fw_policies.removeLockdownWhitelistCommand(command) @handle_exceptions def queryLockdownWhitelistCommand(self, command): return dbus_to_python(self.fw_policies.queryLockdownWhitelistCommand(command)) @handle_exceptions def getLockdownWhitelistCommands(self): return dbus_to_python(self.fw_policies.getLockdownWhitelistCommands()) # context @handle_exceptions def addLockdownWhitelistContext(self, context): self.fw_policies.addLockdownWhitelistContext(context) @handle_exceptions def removeLockdownWhitelistContext(self, context): self.fw_policies.removeLockdownWhitelistContext(context) @handle_exceptions def queryLockdownWhitelistContext(self, context): return dbus_to_python(self.fw_policies.queryLockdownWhitelistContext(context)) @handle_exceptions def getLockdownWhitelistContexts(self): return dbus_to_python(self.fw_policies.getLockdownWhitelistContexts()) # user @handle_exceptions def addLockdownWhitelistUser(self, user): self.fw_policies.addLockdownWhitelistUser(user) @handle_exceptions def removeLockdownWhitelistUser(self, user): self.fw_policies.removeLockdownWhitelistUser(user) @handle_exceptions def queryLockdownWhitelistUser(self, user): return dbus_to_python(self.fw_policies.queryLockdownWhitelistUser(user)) @handle_exceptions def getLockdownWhitelistUsers(self): return dbus_to_python(self.fw_policies.getLockdownWhitelistUsers()) # uid @handle_exceptions def getLockdownWhitelistUids(self): return dbus_to_python(self.fw_policies.getLockdownWhitelistUids()) @handle_exceptions def setLockdownWhitelistUids(self, uids): self.fw_policies.setLockdownWhitelistUids(uids) @handle_exceptions def addLockdownWhitelistUid(self, uid): self.fw_policies.addLockdownWhitelistUid(uid) @handle_exceptions def removeLockdownWhitelistUid(self, uid): self.fw_policies.removeLockdownWhitelistUid(uid) @handle_exceptions def queryLockdownWhitelistUid(self, uid): return dbus_to_python(self.fw_policies.queryLockdownWhitelistUid(uid)) # config.direct class FirewallClientDirect(object): @handle_exceptions def __init__(self, settings=None): if settings: self.settings = settings else: self.settings = [ [], [], [], ] @handle_exceptions def __repr__(self): return '%s(%r)' % (self.__class__, self.settings) @handle_exceptions def getAllChains(self): return self.settings[0] @handle_exceptions def getChains(self, ipv, table): return [ entry[2] for entry in self.settings[0] \ if entry[0] == ipv and entry[1] == table ] @handle_exceptions def setAllChains(self, chains): self.settings[0] = chains @handle_exceptions def addChain(self, ipv, table, chain): idx = (ipv, table, chain) if idx not in self.settings[0]: self.settings[0].append(idx) @handle_exceptions def removeChain(self, ipv, table, chain): idx = (ipv, table, chain) if idx in self.settings[0]: self.settings[0].remove(idx) @handle_exceptions def queryChain(self, ipv, table, chain): idx = (ipv, table, chain) return idx in self.settings[0] @handle_exceptions def getAllRules(self): return self.settings[1] @handle_exceptions def getRules(self, ipv, table, chain): return [ entry[3:] for entry in self.settings[1] \ if entry[0] == ipv and entry[1] == table \ and entry[2] == chain ] @handle_exceptions def setAllRules(self, rules): self.settings[1] = rules @handle_exceptions def addRule(self, ipv, table, chain, priority, args): idx = (ipv, table, chain, priority, args) if idx not in self.settings[1]: self.settings[1].append(idx) @handle_exceptions def removeRule(self, ipv, table, chain, priority, args): idx = (ipv, table, chain, priority, args) if idx in self.settings[1]: self.settings[1].remove(idx) @handle_exceptions def removeRules(self, ipv, table, chain): for idx in list(self.settings[1]): if idx[0] == ipv and idx[1] == table and idx[2] == chain: self.settings[1].remove(idx) @handle_exceptions def queryRule(self, ipv, table, chain, priority, args): idx = (ipv, table, chain, priority, args) return idx in self.settings[1] @handle_exceptions def getAllPassthroughs(self): return self.settings[2] @handle_exceptions def setAllPassthroughs(self, passthroughs): self.settings[2] = passthroughs @handle_exceptions def removeAllPassthroughs(self): self.settings[2] = [] @handle_exceptions def getPassthroughs(self, ipv): return [ entry[1] for entry in self.settings[2] \ if entry[0] == ipv ] @handle_exceptions def addPassthrough(self, ipv, args): idx = (ipv, args) if idx not in self.settings[2]: self.settings[2].append(idx) @handle_exceptions def removePassthrough(self, ipv, args): idx = (ipv, args) if idx in self.settings[2]: self.settings[2].remove(idx) @handle_exceptions def queryPassthrough(self, ipv, args): idx = (ipv, args) return idx in self.settings[2] # config.direct class FirewallClientConfigDirect(object): @handle_exceptions def __init__(self, bus): self.bus = bus self.dbus_obj = self.bus.get_object(config.dbus.DBUS_INTERFACE, config.dbus.DBUS_PATH_CONFIG) self.fw_direct = dbus.Interface( self.dbus_obj, dbus_interface=config.dbus.DBUS_INTERFACE_CONFIG_DIRECT) @handle_exceptions def getSettings(self): return FirewallClientDirect( \ list(dbus_to_python(self.fw_direct.getSettings()))) @handle_exceptions def update(self, settings): self.fw_direct.update(tuple(settings.settings)) # direct chain @handle_exceptions def addChain(self, ipv, table, chain): self.fw_direct.addChain(ipv, table, chain) @handle_exceptions def removeChain(self, ipv, table, chain): self.fw_direct.removeChain(ipv, table, chain) @handle_exceptions def queryChain(self, ipv, table, chain): return dbus_to_python(self.fw_direct.queryChain(ipv, table, chain)) @handle_exceptions def getChains(self, ipv, table): return dbus_to_python(self.fw_direct.getChains(ipv, table)) @handle_exceptions def getAllChains(self): return dbus_to_python(self.fw_direct.getAllChains()) # direct rule @handle_exceptions def addRule(self, ipv, table, chain, priority, args): self.fw_direct.addRule(ipv, table, chain, priority, args) @handle_exceptions def removeRule(self, ipv, table, chain, priority, args): self.fw_direct.removeRule(ipv, table, chain, priority, args) @handle_exceptions def removeRules(self, ipv, table, chain): self.fw_direct.removeRules(ipv, table, chain) @handle_exceptions def queryRule(self, ipv, table, chain, priority, args): return dbus_to_python(self.fw_direct.queryRule(ipv, table, chain, priority, args)) @handle_exceptions def getRules(self, ipv, table, chain): return dbus_to_python(self.fw_direct.getRules(ipv, table, chain)) @handle_exceptions def getAllRules(self): return dbus_to_python(self.fw_direct.getAllRules()) # tracked passthrough @handle_exceptions def addPassthrough(self, ipv, args): self.fw_direct.addPassthrough(ipv, args) @handle_exceptions def removePassthrough(self, ipv, args): self.fw_direct.removePassthrough(ipv, args) @handle_exceptions def queryPassthrough(self, ipv, args): return dbus_to_python(self.fw_direct.queryPassthrough(ipv, args)) @handle_exceptions def getPassthroughs(self, ipv): return dbus_to_python(self.fw_direct.getPassthroughs(ipv)) @handle_exceptions def getAllPassthroughs(self): return dbus_to_python(self.fw_direct.getAllPassthroughs()) # config class FirewallClientConfig(object): @handle_exceptions def __init__(self, bus): self.bus = bus self.dbus_obj = self.bus.get_object(config.dbus.DBUS_INTERFACE, config.dbus.DBUS_PATH_CONFIG) self.fw_config = dbus.Interface( self.dbus_obj, dbus_interface=config.dbus.DBUS_INTERFACE_CONFIG) self.fw_properties = dbus.Interface( self.dbus_obj, dbus_interface='org.freedesktop.DBus.Properties') self._policies = FirewallClientConfigPolicies(self.bus) self._direct = FirewallClientConfigDirect(self.bus) # properties @handle_exceptions def get_property(self, prop): return dbus_to_python(self.fw_properties.Get( config.dbus.DBUS_INTERFACE_CONFIG, prop)) @handle_exceptions def get_properties(self): return dbus_to_python(self.fw_properties.GetAll( config.dbus.DBUS_INTERFACE_CONFIG)) @handle_exceptions def set_property(self, prop, value): self.fw_properties.Set(config.dbus.DBUS_INTERFACE_CONFIG, prop, value) # ipset @handle_exceptions def getIPSetNames(self): return dbus_to_python(self.fw_config.getIPSetNames()) @handle_exceptions def listIPSets(self): return dbus_to_python(self.fw_config.listIPSets()) @handle_exceptions def getIPSet(self, path): return FirewallClientConfigIPSet(self.bus, path) @handle_exceptions def getIPSetByName(self, name): path = dbus_to_python(self.fw_config.getIPSetByName(name)) return FirewallClientConfigIPSet(self.bus, path) @handle_exceptions def addIPSet(self, name, settings): if isinstance(settings, FirewallClientIPSetSettings): path = self.fw_config.addIPSet(name, tuple(settings.settings)) else: path = self.fw_config.addIPSet(name, tuple(settings)) return FirewallClientConfigIPSet(self.bus, path) # zone @handle_exceptions def getZoneNames(self): return dbus_to_python(self.fw_config.getZoneNames()) @handle_exceptions def listZones(self): return dbus_to_python(self.fw_config.listZones()) @handle_exceptions def getZone(self, path): return FirewallClientConfigZone(self.bus, path) @handle_exceptions def getZoneByName(self, name): path = dbus_to_python(self.fw_config.getZoneByName(name)) return FirewallClientConfigZone(self.bus, path) @handle_exceptions def getZoneOfInterface(self, iface): return dbus_to_python(self.fw_config.getZoneOfInterface(iface)) @handle_exceptions def getZoneOfSource(self, source): return dbus_to_python(self.fw_config.getZoneOfSource(source)) @handle_exceptions def addZone(self, name, settings): if isinstance(settings, FirewallClientZoneSettings): path = self.fw_config.addZone2(name, settings.getSettingsDbusDict()) elif isinstance(settings, dict): path = self.fw_config.addZone2(name, settings) else: # tuple based dbus API has 16 elements. Slice what we're given down # to the expected size. path = self.fw_config.addZone(name, tuple(settings[:16])) return FirewallClientConfigZone(self.bus, path) # policy @handle_exceptions def getPolicyNames(self): return dbus_to_python(self.fw_config.getPolicyNames()) @handle_exceptions def listPolicies(self): return dbus_to_python(self.fw_config.listPolicies()) @handle_exceptions def getPolicy(self, path): return FirewallClientConfigPolicy(self.bus, path) @handle_exceptions def getPolicyByName(self, name): path = dbus_to_python(self.fw_config.getPolicyByName(name)) return FirewallClientConfigPolicy(self.bus, path) @handle_exceptions def addPolicy(self, name, settings): if isinstance(settings, FirewallClientPolicySettings): path = self.fw_config.addPolicy(name, settings.getSettingsDbusDict()) else: # dict path = self.fw_config.addPolicy(name, settings) return FirewallClientConfigPolicy(self.bus, path) # service @handle_exceptions def getServiceNames(self): return dbus_to_python(self.fw_config.getServiceNames()) @handle_exceptions def listServices(self): return dbus_to_python(self.fw_config.listServices()) @handle_exceptions def getService(self, path): return FirewallClientConfigService(self.bus, path) @handle_exceptions def getServiceByName(self, name): path = dbus_to_python(self.fw_config.getServiceByName(name)) return FirewallClientConfigService(self.bus, path) @handle_exceptions def addService(self, name, settings): if isinstance(settings, FirewallClientServiceSettings): path = self.fw_config.addService2(name, settings.getSettingsDbusDict()) elif type(settings) is dict: path = self.fw_config.addService2(name, settings) else: # tuple based dbus API has 8 elements. Slice what we're given down # to the expected size. path = self.fw_config.addService(name, tuple(settings[:8])) return FirewallClientConfigService(self.bus, path) # icmptype @handle_exceptions def getIcmpTypeNames(self): return dbus_to_python(self.fw_config.getIcmpTypeNames()) @handle_exceptions def listIcmpTypes(self): return dbus_to_python(self.fw_config.listIcmpTypes()) @handle_exceptions def getIcmpType(self, path): return FirewallClientConfigIcmpType(self.bus, path) @handle_exceptions def getIcmpTypeByName(self, name): path = dbus_to_python(self.fw_config.getIcmpTypeByName(name)) return FirewallClientConfigIcmpType(self.bus, path) @handle_exceptions def addIcmpType(self, name, settings): if isinstance(settings, FirewallClientIcmpTypeSettings): path = self.fw_config.addIcmpType(name, tuple(settings.settings)) else: path = self.fw_config.addIcmpType(name, tuple(settings)) return FirewallClientConfigIcmpType(self.bus, path) @handle_exceptions def policies(self): return self._policies @handle_exceptions def direct(self): return self._direct # helper @handle_exceptions def getHelperNames(self): return dbus_to_python(self.fw_config.getHelperNames()) @handle_exceptions def listHelpers(self): return dbus_to_python(self.fw_config.listHelpers()) @handle_exceptions def getHelper(self, path): return FirewallClientConfigHelper(self.bus, path) @handle_exceptions def getHelperByName(self, name): path = dbus_to_python(self.fw_config.getHelperByName(name)) return FirewallClientConfigHelper(self.bus, path) @handle_exceptions def addHelper(self, name, settings): if isinstance(settings, FirewallClientHelperSettings): path = self.fw_config.addHelper(name, tuple(settings.settings)) else: path = self.fw_config.addHelper(name, tuple(settings)) return FirewallClientConfigHelper(self.bus, path) # class FirewallClient(object): @handle_exceptions def __init__(self, bus=None, wait=0, quiet=True): if not bus: dbus.mainloop.glib.DBusGMainLoop(set_as_default=True) try: self.bus = dbus.SystemBus() except dbus.exceptions.DBusException as e: raise FirewallError(errors.DBUS_ERROR, e.get_dbus_message()) else: self.bus = bus self.bus.add_signal_receiver( handler_function=self._dbus_connection_changed, signal_name="NameOwnerChanged", dbus_interface="org.freedesktop.DBus", arg0=config.dbus.DBUS_INTERFACE) for interface in [ config.dbus.DBUS_INTERFACE, config.dbus.DBUS_INTERFACE_IPSET, config.dbus.DBUS_INTERFACE_ZONE, config.dbus.DBUS_INTERFACE_POLICY, config.dbus.DBUS_INTERFACE_DIRECT, config.dbus.DBUS_INTERFACE_POLICIES, config.dbus.DBUS_INTERFACE_CONFIG, config.dbus.DBUS_INTERFACE_CONFIG_IPSET, config.dbus.DBUS_INTERFACE_CONFIG_ZONE, config.dbus.DBUS_INTERFACE_CONFIG_POLICY, config.dbus.DBUS_INTERFACE_CONFIG_SERVICE, config.dbus.DBUS_INTERFACE_CONFIG_HELPER, config.dbus.DBUS_INTERFACE_CONFIG_DIRECT, config.dbus.DBUS_INTERFACE_CONFIG_ICMPTYPE, config.dbus.DBUS_INTERFACE_CONFIG_POLICIES ]: self.bus.add_signal_receiver(self._signal_receiver, dbus_interface=interface, interface_keyword='interface', member_keyword='member', path_keyword='path') # callbacks self._callback = { } self._callbacks = { # client callbacks "connection-changed": "connection-changed", "connection-established": "connection-established", "connection-lost": "connection-lost", # firewalld callbacks "log-denied-changed": "LogDeniedChanged", "default-zone-changed": "DefaultZoneChanged", "panic-mode-enabled": "PanicModeEnabled", "panic-mode-disabled": "PanicModeDisabled", "reloaded": "Reloaded", "service-added": "ServiceAdded", "service-removed": "ServiceRemoved", "port-added": "PortAdded", "port-removed": "PortRemoved", "source-port-added": "SourcePortAdded", "source-port-removed": "SourcePortRemoved", "protocol-added": "ProtocolAdded", "protocol-removed": "ProtocolRemoved", "masquerade-added": "MasqueradeAdded", "masquerade-removed": "MasqueradeRemoved", "forward-port-added": "ForwardPortAdded", "forward-port-removed": "ForwardPortRemoved", "icmp-block-added": "IcmpBlockAdded", "icmp-block-removed": "IcmpBlockRemoved", "icmp-block-inversion-added": "IcmpBlockInversionAdded", "icmp-block-inversion-removed": "IcmpBlockInversionRemoved", "richrule-added": "RichRuleAdded", "richrule-removed": "RichRuleRemoved", "interface-added": "InterfaceAdded", "interface-removed": "InterfaceRemoved", "zone-changed": "ZoneOfInterfaceChanged", # DEPRECATED, use zone-of-interface-changed instead "zone-of-interface-changed": "ZoneOfInterfaceChanged", "source-added": "SourceAdded", "source-removed": "SourceRemoved", "zone-of-source-changed": "ZoneOfSourceChanged", "zone-updated": "ZoneUpdated", "policy-updated": "PolicyUpdated", # ipset callbacks "ipset-entry-added": "EntryAdded", "ipset-entry-removed": "EntryRemoved", # direct callbacks "direct:chain-added": "ChainAdded", "direct:chain-removed": "ChainRemoved", "direct:rule-added": "RuleAdded", "direct:rule-removed": "RuleRemoved", "direct:passthrough-added": "PassthroughAdded", "direct:passthrough-removed": "PassthroughRemoved", "config:direct:updated": "config:direct:Updated", # policy callbacks "lockdown-enabled": "LockdownEnabled", "lockdown-disabled": "LockdownDisabled", "lockdown-whitelist-command-added": "LockdownWhitelistCommandAdded", "lockdown-whitelist-command-removed": "LockdownWhitelistCommandRemoved", "lockdown-whitelist-context-added": "LockdownWhitelistContextAdded", "lockdown-whitelist-context-removed": "LockdownWhitelistContextRemoved", "lockdown-whitelist-uid-added": "LockdownWhitelistUidAdded", "lockdown-whitelist-uid-removed": "LockdownWhitelistUidRemoved", "lockdown-whitelist-user-added": "LockdownWhitelistUserAdded", "lockdown-whitelist-user-removed": "LockdownWhitelistUserRemoved", # firewalld.config callbacks "config:policies:lockdown-whitelist-updated": "config:policies:LockdownWhitelistUpdated", "config:ipset-added": "config:IPSetAdded", "config:ipset-updated": "config:IPSetUpdated", "config:ipset-removed": "config:IPSetRemoved", "config:ipset-renamed": "config:IPSetRenamed", "config:zone-added": "config:ZoneAdded", "config:zone-updated": "config:ZoneUpdated", "config:zone-removed": "config:ZoneRemoved", "config:zone-renamed": "config:ZoneRenamed", "config:policy-added": "config:PolicyAdded", "config:policy-updated": "config:PolicyUpdated", "config:policy-removed": "config:PolicyRemoved", "config:policy-renamed": "config:PolicyRenamed", "config:service-added": "config:ServiceAdded", "config:service-updated": "config:ServiceUpdated", "config:service-removed": "config:ServiceRemoved", "config:service-renamed": "config:ServiceRenamed", "config:icmptype-added": "config:IcmpTypeAdded", "config:icmptype-updated": "config:IcmpTypeUpdated", "config:icmptype-removed": "config:IcmpTypeRemoved", "config:icmptype-renamed": "config:IcmpTypeRenamed", "config:helper-added": "config:HelperAdded", "config:helper-updated": "config:HelperUpdated", "config:helper-removed": "config:HelperRemoved", "config:helper-renamed": "config:HelperRenamed", } # initialize variables used for connection self._init_vars() self.quiet = quiet if wait > 0: # connect in one second GLib.timeout_add_seconds(wait, self._connection_established) else: self._connection_established() @handle_exceptions def _init_vars(self): self.fw = None self.fw_ipset = None self.fw_zone = None self.fw_policy = None self.fw_helper = None self.fw_direct = None self.fw_properties = None self._config = None self.connected = False @handle_exceptions def getExceptionHandler(self): return exception_handler @handle_exceptions def setExceptionHandler(self, handler): global exception_handler exception_handler = handler @handle_exceptions def getNotAuthorizedLoop(self): return not_authorized_loop @handle_exceptions def setNotAuthorizedLoop(self, enable): global not_authorized_loop not_authorized_loop = enable @handle_exceptions def connect(self, name, callback, args): if name in self._callbacks: self._callback[self._callbacks[name]] = (callback, args) else: raise ValueError("Unknown callback name '%s'" % name) @handle_exceptions def _dbus_connection_changed(self, name, old_owner, new_owner): if name != config.dbus.DBUS_INTERFACE: return if new_owner: # connection established self._connection_established() else: # connection lost self._connection_lost() @handle_exceptions def _connection_established(self): try: self.dbus_obj = self.bus.get_object(config.dbus.DBUS_INTERFACE, config.dbus.DBUS_PATH) self.fw = dbus.Interface(self.dbus_obj, dbus_interface=config.dbus.DBUS_INTERFACE) self.fw_ipset = dbus.Interface( self.dbus_obj, dbus_interface=config.dbus.DBUS_INTERFACE_IPSET) self.fw_zone = dbus.Interface( self.dbus_obj, dbus_interface=config.dbus.DBUS_INTERFACE_ZONE) self.fw_policy = dbus.Interface( self.dbus_obj, dbus_interface=config.dbus.DBUS_INTERFACE_POLICY) self.fw_direct = dbus.Interface( self.dbus_obj, dbus_interface=config.dbus.DBUS_INTERFACE_DIRECT) self.fw_policies = dbus.Interface( self.dbus_obj, dbus_interface=config.dbus.DBUS_INTERFACE_POLICIES) self.fw_properties = dbus.Interface( self.dbus_obj, dbus_interface='org.freedesktop.DBus.Properties') except dbus.exceptions.DBusException as e: # ignore dbus errors if not self.quiet: print ("DBusException", e.get_dbus_message()) return except Exception as e: if not self.quiet: print ("Exception", e) return self._config = FirewallClientConfig(self.bus) self.connected = True self._signal_receiver(member="connection-established", interface=config.dbus.DBUS_INTERFACE) self._signal_receiver(member="connection-changed", interface=config.dbus.DBUS_INTERFACE) @handle_exceptions def _connection_lost(self): self._init_vars() self._signal_receiver(member="connection-lost", interface=config.dbus.DBUS_INTERFACE) self._signal_receiver(member="connection-changed", interface=config.dbus.DBUS_INTERFACE) @handle_exceptions def _signal_receiver(self, args, kwargs): if "member" not in kwargs or "interface" not in kwargs: return signal = kwargs["member"] interface = kwargs["interface"] # config signals need special treatment # pimp signal name if interface.startswith(config.dbus.DBUS_INTERFACE_CONFIG_ZONE): signal = "config:Zone" + signal if interface.startswith(config.dbus.DBUS_INTERFACE_CONFIG_POLICY): signal = "config:Policy" + signal elif interface.startswith(config.dbus.DBUS_INTERFACE_CONFIG_IPSET): signal = "config:IPSet" + signal elif interface.startswith(config.dbus.DBUS_INTERFACE_CONFIG_SERVICE): signal = "config:Service" + signal elif interface.startswith(config.dbus.DBUS_INTERFACE_CONFIG_ICMPTYPE): signal = "config:IcmpType" + signal elif interface.startswith(config.dbus.DBUS_INTERFACE_CONFIG_HELPER): signal = "config:Helper" + signal elif interface == config.dbus.DBUS_INTERFACE_CONFIG: signal = "config:" + signal elif interface == config.dbus.DBUS_INTERFACE_CONFIG_POLICIES: signal = "config:policies:" + signal elif interface == config.dbus.DBUS_INTERFACE_CONFIG_DIRECT: signal = "config:direct:" + signal cb = None for callback in self._callbacks: if self._callbacks[callback] == signal and \ self._callbacks[callback] in self._callback: cb = self._callback[self._callbacks[callback]] if cb is None: return # call back with args converted to python types ... cb_args = [ dbus_to_python(arg) for arg in args ] try: if cb[1]: # add call data cb_args.extend(cb[1]) # call back cb[0](cb_args) except Exception as msg: print(msg) @handle_exceptions def config(self): return self._config @handle_exceptions def reload(self): self.fw.reload() @handle_exceptions def resetToDefaults(self): self.fw.resetToDefaults() @handle_exceptions def complete_reload(self): self.fw.completeReload() @handle_exceptions def runtimeToPermanent(self): self.fw.runtimeToPermanent() @handle_exceptions def checkPermanentConfig(self): self.fw.checkPermanentConfig() @handle_exceptions def get_property(self, prop): return dbus_to_python(self.fw_properties.Get( config.dbus.DBUS_INTERFACE, prop)) @handle_exceptions def get_properties(self): return dbus_to_python(self.fw_properties.GetAll( config.dbus.DBUS_INTERFACE)) @handle_exceptions def set_property(self, prop, value): self.fw_properties.Set(config.dbus.DBUS_INTERFACE, prop, value) # panic mode @handle_exceptions def enablePanicMode(self): self.fw.enablePanicMode() @handle_exceptions def disablePanicMode(self): self.fw.disablePanicMode() @handle_exceptions def queryPanicMode(self): return dbus_to_python(self.fw.queryPanicMode()) # list functions @handle_exceptions def getZoneSettings(self, zone): return FirewallClientZoneSettings(dbus_to_python(self.fw_zone.getZoneSettings2(zone))) @handle_exceptions def getIPSets(self): return dbus_to_python(self.fw_ipset.getIPSets()) @handle_exceptions def getIPSetSettings(self, ipset): return FirewallClientIPSetSettings(list(dbus_to_python(\ self.fw_ipset.getIPSetSettings(ipset)))) @handle_exceptions def addEntry(self, ipset, entry): self.fw_ipset.addEntry(ipset, entry) @handle_exceptions def getEntries(self, ipset): return self.fw_ipset.getEntries(ipset) @handle_exceptions def setEntries(self, ipset, entries): return self.fw_ipset.setEntries(ipset, entries) @handle_exceptions def removeEntry(self, ipset, entry): self.fw_ipset.removeEntry(ipset, entry) @handle_exceptions def queryEntry(self, ipset, entry): return dbus_to_python(self.fw_ipset.queryEntry(ipset, entry)) @handle_exceptions def listServices(self): return dbus_to_python(self.fw.listServices()) @handle_exceptions def getServiceSettings(self, service): return FirewallClientServiceSettings(dbus_to_python( self.fw.getServiceSettings2(service))) @handle_exceptions def listIcmpTypes(self): return dbus_to_python(self.fw.listIcmpTypes()) @handle_exceptions def getIcmpTypeSettings(self, icmptype): return FirewallClientIcmpTypeSettings(list(dbus_to_python(\ self.fw.getIcmpTypeSettings(icmptype)))) @handle_exceptions def getHelpers(self): return dbus_to_python(self.fw.getHelpers()) @handle_exceptions def getHelperSettings(self, helper): return FirewallClientHelperSettings(list(dbus_to_python(\ self.fw.getHelperSettings(helper)))) # automatic helper setting @handle_exceptions def getAutomaticHelpers(self): return dbus_to_python(self.fw.getAutomaticHelpers()) @handle_exceptions def setAutomaticHelpers(self, value): self.fw.setAutomaticHelpers(value) # log denied @handle_exceptions def getLogDenied(self): return dbus_to_python(self.fw.getLogDenied()) @handle_exceptions def setLogDenied(self, value): self.fw.setLogDenied(value) # default zone @handle_exceptions def getDefaultZone(self): return dbus_to_python(self.fw.getDefaultZone()) @handle_exceptions def setDefaultZone(self, zone): self.fw.setDefaultZone(zone) # zone @handle_exceptions def setZoneSettings(self, zone, settings): self.fw_zone.setZoneSettings2(zone, settings.getRuntimeSettingsDbusDict()) @handle_exceptions def getZones(self): return dbus_to_python(self.fw_zone.getZones()) @handle_exceptions def getActiveZones(self): return dbus_to_python(self.fw_zone.getActiveZones()) @handle_exceptions def getZoneOfInterface(self, interface): return dbus_to_python(self.fw_zone.getZoneOfInterface(interface)) @handle_exceptions def getZoneOfSource(self, source): return dbus_to_python(self.fw_zone.getZoneOfSource(source)) @handle_exceptions def isImmutable(self, zone): return dbus_to_python(self.fw_zone.isImmutable(zone)) # policy @handle_exceptions def getPolicySettings(self, policy): return FirewallClientPolicySettings(dbus_to_python(self.fw_policy.getPolicySettings(policy))) @handle_exceptions def setPolicySettings(self, policy, settings): self.fw_policy.setPolicySettings(policy, settings.getRuntimeSettingsDbusDict()) @handle_exceptions def getPolicies(self): return dbus_to_python(self.fw_policy.getPolicies()) @handle_exceptions def getActivePolicies(self): return dbus_to_python(self.fw_policy.getActivePolicies()) @handle_exceptions def isPolicyImmutable(self, policy): return dbus_to_python(self.fw_policy.isImmutable(policy)) # interfaces @handle_exceptions def addInterface(self, zone, interface): return dbus_to_python(self.fw_zone.addInterface(zone, interface)) @handle_exceptions def changeZone(self, zone, interface): # DEPRECATED return dbus_to_python(self.fw_zone.changeZone(zone, interface)) @handle_exceptions def changeZoneOfInterface(self, zone, interface): return dbus_to_python(self.fw_zone.changeZoneOfInterface(zone, interface)) @handle_exceptions def getInterfaces(self, zone): return dbus_to_python(self.fw_zone.getInterfaces(zone)) @handle_exceptions def queryInterface(self, zone, interface): return dbus_to_python(self.fw_zone.queryInterface(zone, interface)) @handle_exceptions def removeInterface(self, zone, interface): return dbus_to_python(self.fw_zone.removeInterface(zone, interface)) # sources @handle_exceptions def addSource(self, zone, source): return dbus_to_python(self.fw_zone.addSource(zone, source)) @handle_exceptions def changeZoneOfSource(self, zone, source): return dbus_to_python(self.fw_zone.changeZoneOfSource(zone, source)) @handle_exceptions def getSources(self, zone): return dbus_to_python(self.fw_zone.getSources(zone)) @handle_exceptions def querySource(self, zone, source): return dbus_to_python(self.fw_zone.querySource(zone, source)) @handle_exceptions def removeSource(self, zone, source): return dbus_to_python(self.fw_zone.removeSource(zone, source)) # rich rules @handle_exceptions def addRichRule(self, zone, rule, timeout=0): return dbus_to_python(self.fw_zone.addRichRule(zone, rule, timeout)) @handle_exceptions def getRichRules(self, zone): return dbus_to_python(self.fw_zone.getRichRules(zone)) @handle_exceptions def queryRichRule(self, zone, rule): return dbus_to_python(self.fw_zone.queryRichRule(zone, rule)) @handle_exceptions def removeRichRule(self, zone, rule): return dbus_to_python(self.fw_zone.removeRichRule(zone, rule)) # services @handle_exceptions def addService(self, zone, service, timeout=0): return dbus_to_python(self.fw_zone.addService(zone, service, timeout)) @handle_exceptions def getServices(self, zone): return dbus_to_python(self.fw_zone.getServices(zone)) @handle_exceptions def queryService(self, zone, service): return dbus_to_python(self.fw_zone.queryService(zone, service)) @handle_exceptions def removeService(self, zone, service): return dbus_to_python(self.fw_zone.removeService(zone, service)) # ports @handle_exceptions def addPort(self, zone, port, protocol, timeout=0): return dbus_to_python(self.fw_zone.addPort(zone, port, protocol, timeout)) @handle_exceptions def getPorts(self, zone): return dbus_to_python(self.fw_zone.getPorts(zone)) @handle_exceptions def queryPort(self, zone, port, protocol): return dbus_to_python(self.fw_zone.queryPort(zone, port, protocol)) @handle_exceptions def removePort(self, zone, port, protocol): return dbus_to_python(self.fw_zone.removePort(zone, port, protocol)) # protocols @handle_exceptions def addProtocol(self, zone, protocol, timeout=0): return dbus_to_python(self.fw_zone.addProtocol(zone, protocol, timeout)) @handle_exceptions def getProtocols(self, zone): return dbus_to_python(self.fw_zone.getProtocols(zone)) @handle_exceptions def queryProtocol(self, zone, protocol): return dbus_to_python(self.fw_zone.queryProtocol(zone, protocol)) @handle_exceptions def removeProtocol(self, zone, protocol): return dbus_to_python(self.fw_zone.removeProtocol(zone, protocol)) # forward @handle_exceptions def addForward(self, zone): self.fw_zone.setZoneSettings2(zone, {"forward": True}) @handle_exceptions def queryForward(self, zone): return dbus_to_python(self.fw_zone.getZoneSettings2(zone))["forward"] @handle_exceptions def removeForward(self, zone): self.fw_zone.setZoneSettings2(zone, {"forward": False}) # masquerade @handle_exceptions def addMasquerade(self, zone, timeout=0): return dbus_to_python(self.fw_zone.addMasquerade(zone, timeout)) @handle_exceptions def queryMasquerade(self, zone): return dbus_to_python(self.fw_zone.queryMasquerade(zone)) @handle_exceptions def removeMasquerade(self, zone): return dbus_to_python(self.fw_zone.removeMasquerade(zone)) # forward ports @handle_exceptions def addForwardPort(self, zone, port, protocol, toport, toaddr, timeout=0): if toport is None: toport = "" if toaddr is None: toaddr = "" return dbus_to_python(self.fw_zone.addForwardPort(zone, port, protocol, toport, toaddr, timeout)) @handle_exceptions def getForwardPorts(self, zone): return dbus_to_python(self.fw_zone.getForwardPorts(zone)) @handle_exceptions def queryForwardPort(self, zone, port, protocol, toport, toaddr): if toport is None: toport = "" if toaddr is None: toaddr = "" return dbus_to_python(self.fw_zone.queryForwardPort(zone, port, protocol, toport, toaddr)) @handle_exceptions def removeForwardPort(self, zone, port, protocol, toport, toaddr): if toport is None: toport = "" if toaddr is None: toaddr = "" return dbus_to_python(self.fw_zone.removeForwardPort(zone, port, protocol, toport, toaddr)) # source ports @handle_exceptions def addSourcePort(self, zone, port, protocol, timeout=0): return dbus_to_python(self.fw_zone.addSourcePort(zone, port, protocol, timeout)) @handle_exceptions def getSourcePorts(self, zone): return dbus_to_python(self.fw_zone.getSourcePorts(zone)) @handle_exceptions def querySourcePort(self, zone, port, protocol): return dbus_to_python(self.fw_zone.querySourcePort(zone, port, protocol)) @handle_exceptions def removeSourcePort(self, zone, port, protocol): return dbus_to_python(self.fw_zone.removeSourcePort(zone, port, protocol)) # icmpblock @handle_exceptions def addIcmpBlock(self, zone, icmp, timeout=0): return dbus_to_python(self.fw_zone.addIcmpBlock(zone, icmp, timeout)) @handle_exceptions def getIcmpBlocks(self, zone): return dbus_to_python(self.fw_zone.getIcmpBlocks(zone)) @handle_exceptions def queryIcmpBlock(self, zone, icmp): return dbus_to_python(self.fw_zone.queryIcmpBlock(zone, icmp)) @handle_exceptions def removeIcmpBlock(self, zone, icmp): return dbus_to_python(self.fw_zone.removeIcmpBlock(zone, icmp)) # icmp block inversion @handle_exceptions def addIcmpBlockInversion(self, zone): return dbus_to_python(self.fw_zone.addIcmpBlockInversion(zone)) @handle_exceptions def queryIcmpBlockInversion(self, zone): return dbus_to_python(self.fw_zone.queryIcmpBlockInversion(zone)) @handle_exceptions def removeIcmpBlockInversion(self, zone): return dbus_to_python(self.fw_zone.removeIcmpBlockInversion(zone)) # direct chain @handle_exceptions def addChain(self, ipv, table, chain): self.fw_direct.addChain(ipv, table, chain) @handle_exceptions def removeChain(self, ipv, table, chain): self.fw_direct.removeChain(ipv, table, chain) @handle_exceptions def queryChain(self, ipv, table, chain): return dbus_to_python(self.fw_direct.queryChain(ipv, table, chain)) @handle_exceptions def getChains(self, ipv, table): return dbus_to_python(self.fw_direct.getChains(ipv, table)) @handle_exceptions def getAllChains(self): return dbus_to_python(self.fw_direct.getAllChains()) # direct rule @handle_exceptions def addRule(self, ipv, table, chain, priority, args): self.fw_direct.addRule(ipv, table, chain, priority, args) @handle_exceptions def removeRule(self, ipv, table, chain, priority, args): self.fw_direct.removeRule(ipv, table, chain, priority, args) @handle_exceptions def removeRules(self, ipv, table, chain): self.fw_direct.removeRules(ipv, table, chain) @handle_exceptions def queryRule(self, ipv, table, chain, priority, args): return dbus_to_python(self.fw_direct.queryRule(ipv, table, chain, priority, args)) @handle_exceptions def getRules(self, ipv, table, chain): return dbus_to_python(self.fw_direct.getRules(ipv, table, chain)) @handle_exceptions def getAllRules(self): return dbus_to_python(self.fw_direct.getAllRules()) # direct passthrough @handle_exceptions def passthrough(self, ipv, args): return dbus_to_python(self.fw_direct.passthrough(ipv, args)) # tracked passthrough @handle_exceptions def getAllPassthroughs(self): return dbus_to_python(self.fw_direct.getAllPassthroughs()) @handle_exceptions def removeAllPassthroughs(self): self.fw_direct.removeAllPassthroughs() @handle_exceptions def getPassthroughs(self, ipv): return dbus_to_python(self.fw_direct.getPassthroughs(ipv)) @handle_exceptions def addPassthrough(self, ipv, args): self.fw_direct.addPassthrough(ipv, args) @handle_exceptions def removePassthrough(self, ipv, args): self.fw_direct.removePassthrough(ipv, args) @handle_exceptions def queryPassthrough(self, ipv, args): return dbus_to_python(self.fw_direct.queryPassthrough(ipv, args)) # lockdown @handle_exceptions def enableLockdown(self): self.fw_policies.enableLockdown() @handle_exceptions def disableLockdown(self): self.fw_policies.disableLockdown() @handle_exceptions def queryLockdown(self): return dbus_to_python(self.fw_policies.queryLockdown()) # policies # lockdown white list commands @handle_exceptions def addLockdownWhitelistCommand(self, command): self.fw_policies.addLockdownWhitelistCommand(command) @handle_exceptions def getLockdownWhitelistCommands(self): return dbus_to_python(self.fw_policies.getLockdownWhitelistCommands()) @handle_exceptions def queryLockdownWhitelistCommand(self, command): return dbus_to_python(self.fw_policies.queryLockdownWhitelistCommand(command)) @handle_exceptions def removeLockdownWhitelistCommand(self, command): self.fw_policies.removeLockdownWhitelistCommand(command) # lockdown white list contexts @handle_exceptions def addLockdownWhitelistContext(self, context): self.fw_policies.addLockdownWhitelistContext(context) @handle_exceptions def getLockdownWhitelistContexts(self): return dbus_to_python(self.fw_policies.getLockdownWhitelistContexts()) @handle_exceptions def queryLockdownWhitelistContext(self, context): return dbus_to_python(self.fw_policies.queryLockdownWhitelistContext(context)) @handle_exceptions def removeLockdownWhitelistContext(self, context): self.fw_policies.removeLockdownWhitelistContext(context) # lockdown white list uids @handle_exceptions def addLockdownWhitelistUid(self, uid): self.fw_policies.addLockdownWhitelistUid(uid) @handle_exceptions def getLockdownWhitelistUids(self): return dbus_to_python(self.fw_policies.getLockdownWhitelistUids()) @handle_exceptions def queryLockdownWhitelistUid(self, uid): return dbus_to_python(self.fw_policies.queryLockdownWhitelistUid(uid)) @handle_exceptions def removeLockdownWhitelistUid(self, uid): self.fw_policies.removeLockdownWhitelistUid(uid) # lockdown white list users @handle_exceptions def addLockdownWhitelistUser(self, user): self.fw_policies.addLockdownWhitelistUser(user) @handle_exceptions def getLockdownWhitelistUsers(self): return dbus_to_python(self.fw_policies.getLockdownWhitelistUsers()) @handle_exceptions def queryLockdownWhitelistUser(self, user): return dbus_to_python(self.fw_policies.queryLockdownWhitelistUser(user)) @handle_exceptions def removeLockdownWhitelistUser(self, user): self.fw_policies.removeLockdownWhitelistUser(user) @handle_exceptions def authorizeAll(self): """ Authorize once for all polkit actions. """ self.fw.authorizeAll()	118.52 KB	2026-03-10 23:36
📄 command.py # -- coding: utf-8 -- # # Copyright (C) 2011-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/>. # """FirewallCommand class for command line client simplification""" __all__ = [ "FirewallCommand" ] import sys from firewall import errors from firewall.errors import FirewallError from dbus.exceptions import DBusException from firewall.functions import checkIPnMask, checkIP6nMask, check_mac, \ check_port, check_single_address class FirewallCommand(object): def __init__(self, quiet=False, verbose=False): self.quiet = quiet self.verbose = verbose self.__use_exception_handler = True self.fw = None def set_fw(self, fw): self.fw = fw def set_quiet(self, flag): self.quiet = flag def get_quiet(self): return self.quiet def set_verbose(self, flag): self.verbose = flag def get_verbose(self): return self.verbose def print_msg(self, msg=None): if msg is not None and not self.quiet: sys.stdout.write(msg + "\n") def print_error_msg(self, msg=None): if msg is not None and not self.quiet: sys.stderr.write(msg + "\n") def print_warning(self, msg=None): FAIL = '\033[91m' END = '\033[00m' if sys.stderr.isatty(): msg = FAIL + msg + END self.print_error_msg(msg) def print_and_exit(self, msg=None, exit_code=0): #OK = '\033[92m' #END = '\033[00m' if exit_code > 1: self.print_warning(msg) else: #if sys.stdout.isatty(): # msg = OK + msg + END self.print_msg(msg) sys.exit(exit_code) def fail(self, msg=None): self.print_and_exit(msg, 2) def print_if_verbose(self, msg=None): if msg is not None and self.verbose: sys.stdout.write(msg + "\n") def __cmd_sequence(self, cmd_type, option, action_method, query_method, # pylint: disable=W0613, R0913, R0914 parse_method, message, start_args=None, end_args=None, # pylint: disable=W0613 no_exit=False): if self.fw is not None: self.fw.authorizeAll() items = [ ] _errors = 0 _error_codes = [ ] for item in option: if parse_method is not None: try: item = parse_method(item) except Exception as msg: code = FirewallError.get_code(str(msg)) if len(option) > 1: self.print_warning("Warning: %s" % msg) else: self.print_and_exit("Error: %s" % msg, code) if code not in _error_codes: _error_codes.append(code) _errors += 1 continue items.append(item) for item in items: call_item = [ ] if start_args is not None: call_item += start_args if not isinstance(item, list) and not isinstance(item, tuple): call_item.append(item) else: call_item += item if end_args is not None: call_item += end_args self.deactivate_exception_handler() try: action_method(call_item) except (DBusException, Exception) as msg: if isinstance(msg, DBusException): self.fail_if_not_authorized(msg.get_dbus_name()) msg = msg.get_dbus_message() else: msg = str(msg) code = FirewallError.get_code(msg) if code in [ errors.ALREADY_ENABLED, errors.NOT_ENABLED, errors.ZONE_ALREADY_SET, errors.ALREADY_SET ]: code = 0 if len(option) > 1: self.print_warning("Warning: %s" % msg) elif code == 0: self.print_warning("Warning: %s" % msg) return else: self.print_and_exit("Error: %s" % msg, code) if code not in _error_codes: _error_codes.append(code) _errors += 1 self.activate_exception_handler() if not no_exit: if len(option) > _errors or 0 in _error_codes: # There have been more options than errors or there # was at least one error code 0, return. return elif len(_error_codes) == 1: # Exactly one error code, use it. sys.exit(_error_codes[0]) elif len(_error_codes) > 1: # There is more than error, exit using # UNKNOWN_ERROR. This could happen within sequences # where parsing failed with different errors like # INVALID_PORT and INVALID_PROTOCOL. sys.exit(errors.UNKNOWN_ERROR) def add_sequence(self, option, action_method, query_method, parse_method, # pylint: disable=R0913 message, no_exit=False): self.__cmd_sequence("add", option, action_method, query_method, parse_method, message, no_exit=no_exit) def x_add_sequence(self, x, option, action_method, query_method, # pylint: disable=R0913 parse_method, message, no_exit=False): self.__cmd_sequence("add", option, action_method, query_method, parse_method, message, start_args=[x], no_exit=no_exit) def zone_add_timeout_sequence(self, zone, option, action_method, # pylint: disable=R0913 query_method, parse_method, message, timeout, no_exit=False): self.__cmd_sequence("add", option, action_method, query_method, parse_method, message, start_args=[zone], end_args=[timeout], no_exit=no_exit) def remove_sequence(self, option, action_method, query_method, # pylint: disable=R0913 parse_method, message, no_exit=False): self.__cmd_sequence("remove", option, action_method, query_method, parse_method, message, no_exit=no_exit) def x_remove_sequence(self, x, option, action_method, query_method, # pylint: disable=R0913 parse_method, message, no_exit=False): self.__cmd_sequence("remove", option, action_method, query_method, parse_method, message, start_args=[x], no_exit=no_exit) def __query_sequence(self, option, query_method, parse_method, message, # pylint: disable=R0913 start_args=None, no_exit=False): items = [ ] for item in option: if parse_method is not None: try: item = parse_method(item) except Exception as msg: if len(option) > 1: self.print_warning("Warning: %s" % msg) continue else: code = FirewallError.get_code(str(msg)) self.print_and_exit("Error: %s" % msg, code) items.append(item) for item in items: call_item = [ ] if start_args is not None: call_item += start_args if not isinstance(item, list) and not isinstance(item, tuple): call_item.append(item) else: call_item += item self.deactivate_exception_handler() try: res = query_method(call_item) except DBusException as msg: self.fail_if_not_authorized(msg.get_dbus_name()) code = FirewallError.get_code(msg.get_dbus_message()) if len(option) > 1: self.print_warning("Warning: %s" % msg.get_dbus_message()) continue else: self.print_and_exit("Error: %s" % msg.get_dbus_message(), code) except Exception as msg: code = FirewallError.get_code(str(msg)) if len(option) > 1: self.print_warning("Warning: %s" % msg) else: self.print_and_exit("Error: %s" % msg, code) self.activate_exception_handler() if len(option) > 1: self.print_msg("%s: %s" % (message % item, ("no", "yes")[res])) else: self.print_query_result(res) if not no_exit: sys.exit(0) def query_sequence(self, option, query_method, parse_method, message, # pylint: disable=R0913 no_exit=False): self.__query_sequence(option, query_method, parse_method, message, no_exit=no_exit) def x_query_sequence(self, x, option, query_method, parse_method, # pylint: disable=R0913 message, no_exit=False): self.__query_sequence(option, query_method, parse_method, message, start_args=[x], no_exit=no_exit) def parse_source(self, value): if not checkIPnMask(value) and not checkIP6nMask(value) \ and not check_mac(value) and not \ (value.startswith("ipset:") and len(value) > 6): raise FirewallError(errors.INVALID_ADDR, "'%s' is no valid IPv4, IPv6 or MAC address, nor an ipset" % value) return value def parse_port(self, value, separator="/"): try: (port, proto) = value.split(separator) except ValueError: raise FirewallError(errors.INVALID_PORT, "bad port (most likely " "missing protocol), correct syntax is " "portid[-portid]%sprotocol" % separator) if not check_port(port): raise FirewallError(errors.INVALID_PORT, port) if proto not in [ "tcp", "udp", "sctp", "dccp" ]: raise FirewallError(errors.INVALID_PROTOCOL, "'%s' not in {'tcp'\|'udp'\|'sctp'\|'dccp'}" % \ proto) return (port, proto) def parse_forward_port(self, value, compat=False): port = None protocol = None toport = None toaddr = None i = 0 while ("=" in value[i:]): opt = value[i:].split("=", 1)[0] i += len(opt) + 1 if "=" in value[i:]: val = value[i:].split(":", 1)[0] else: val = value[i:] i += len(val) + 1 if opt == "port": port = val elif opt == "proto": protocol = val elif opt == "toport": toport = val elif opt == "toaddr": toaddr = val elif opt == "if" and compat: # ignore if option in compat mode pass else: raise FirewallError(errors.INVALID_FORWARD, "invalid forward port arg '%s'" % (opt)) if not port: raise FirewallError(errors.INVALID_FORWARD, "missing port") if not protocol: raise FirewallError(errors.INVALID_FORWARD, "missing protocol") if not (toport or toaddr): raise FirewallError(errors.INVALID_FORWARD, "missing destination") if not check_port(port): raise FirewallError(errors.INVALID_PORT, port) if protocol not in [ "tcp", "udp", "sctp", "dccp" ]: raise FirewallError(errors.INVALID_PROTOCOL, "'%s' not in {'tcp'\|'udp'\|'sctp'\|'dccp'}" % \ protocol) if toport and not check_port(toport): raise FirewallError(errors.INVALID_PORT, toport) if toaddr and not check_single_address("ipv4", toaddr): if compat or not check_single_address("ipv6", toaddr): raise FirewallError(errors.INVALID_ADDR, toaddr) return (port, protocol, toport, toaddr) def parse_ipset_option(self, value): args = value.split("=") if len(args) == 1: return (args[0], "") elif len(args) == 2: return args else: raise FirewallError(errors.INVALID_OPTION, "invalid ipset option '%s'" % (value)) def check_destination_ipv(self, value): ipvs = [ "ipv4", "ipv6", ] if value not in ipvs: raise FirewallError(errors.INVALID_IPV, "invalid argument: %s (choose from '%s')" % \ (value, "', '".join(ipvs))) return value def parse_service_destination(self, value): try: (ipv, destination) = value.split(":", 1) except ValueError: raise FirewallError(errors.INVALID_DESTINATION, "destination syntax is ipv:address[/mask]") return (self.check_destination_ipv(ipv), destination) def check_ipv(self, value): ipvs = [ "ipv4", "ipv6", "eb" ] if value not in ipvs: raise FirewallError(errors.INVALID_IPV, "invalid argument: %s (choose from '%s')" % \ (value, "', '".join(ipvs))) return value def check_helper_family(self, value): ipvs = [ "", "ipv4", "ipv6" ] if value not in ipvs: raise FirewallError(errors.INVALID_IPV, "invalid argument: %s (choose from '%s')" % \ (value, "', '".join(ipvs))) return value def check_module(self, value): if not value.startswith("nf_conntrack_"): raise FirewallError( errors.INVALID_MODULE, "'%s' does not start with 'nf_conntrack_'" % value) if len(value.replace("nf_conntrack_", "")) < 1: raise FirewallError(errors.INVALID_MODULE, "Module name '%s' too short" % value) return value def print_zone_policy_info(self, zone, settings, default_zone=None, extra_interfaces=[], isPolicy=True): # pylint: disable=R0914 target = settings.getTarget() services = settings.getServices() ports = settings.getPorts() protocols = settings.getProtocols() masquerade = settings.getMasquerade() forward_ports = settings.getForwardPorts() source_ports = settings.getSourcePorts() icmp_blocks = settings.getIcmpBlocks() rules = settings.getRichRules() description = settings.getDescription() short_description = settings.getShort() if isPolicy: ingress_zones = settings.getIngressZones() egress_zones = settings.getEgressZones() priority = settings.getPriority() else: icmp_block_inversion = settings.getIcmpBlockInversion() interfaces = sorted(set(settings.getInterfaces() + extra_interfaces)) sources = settings.getSources() forward = settings.getForward() def rich_rule_sorted_key(rule): priority = 0 search_str = "priority=" try: i = rule.index(search_str) except ValueError: pass else: i += len(search_str) priority = int(rule[i:i+(rule[i:].index(" "))].replace("\"", "")) return priority attributes = [] if default_zone is not None: if zone == default_zone: attributes.append("default") if (not isPolicy and (interfaces or sources)) or \ ( isPolicy and ingress_zones and egress_zones): attributes.append("active") if attributes: zone = zone + " (%s)" % ", ".join(attributes) self.print_msg(zone) if self.verbose: self.print_msg(" summary: " + short_description) self.print_msg(" description: " + description) if isPolicy: self.print_msg(" priority: " + str(priority)) self.print_msg(" target: " + target) if not isPolicy: self.print_msg(" icmp-block-inversion: %s" % \ ("yes" if icmp_block_inversion else "no")) if isPolicy: self.print_msg(" ingress-zones: " + " ".join(ingress_zones)) self.print_msg(" egress-zones: " + " ".join(egress_zones)) else: self.print_msg(" interfaces: " + " ".join(interfaces)) self.print_msg(" sources: " + " ".join(sources)) self.print_msg(" services: " + " ".join(sorted(services))) self.print_msg(" ports: " + " ".join(["%s/%s" % (port[0], port[1]) for port in ports])) self.print_msg(" protocols: " + " ".join(sorted(protocols))) if not isPolicy: self.print_msg(" forward: %s" % ("yes" if forward else "no")) self.print_msg(" masquerade: %s" % ("yes" if masquerade else "no")) self.print_msg(" forward-ports: " + ("\n\t" if forward_ports else "") + "\n\t".join(["port=%s:proto=%s:toport=%s:toaddr=%s" % \ (port, proto, toport, toaddr) for (port, proto, toport, toaddr) in \ forward_ports])) self.print_msg(" source-ports: " + " ".join(["%s/%s" % (port[0], port[1]) for port in source_ports])) self.print_msg(" icmp-blocks: " + " ".join(icmp_blocks)) self.print_msg(" rich rules: " + ("\n\t" if rules else "") + "\n\t".join(sorted(rules, key=rich_rule_sorted_key))) def print_zone_info(self, zone, settings, default_zone=None, extra_interfaces=[]): self.print_zone_policy_info(zone, settings, default_zone=default_zone, extra_interfaces=extra_interfaces, isPolicy=False) def print_policy_info(self, policy, settings, default_zone=None, extra_interfaces=[]): self.print_zone_policy_info(policy, settings, default_zone=default_zone, extra_interfaces=extra_interfaces, isPolicy=True) def print_service_info(self, service, settings): ports = settings.getPorts() protocols = settings.getProtocols() source_ports = settings.getSourcePorts() modules = settings.getModules() description = settings.getDescription() destinations = settings.getDestinations() short_description = settings.getShort() includes = settings.getIncludes() helpers = settings.getHelpers() self.print_msg(service) if self.verbose: self.print_msg(" summary: " + short_description) self.print_msg(" description: " + description) self.print_msg(" ports: " + " ".join(["%s/%s" % (port[0], port[1]) for port in ports])) self.print_msg(" protocols: " + " ".join(protocols)) self.print_msg(" source-ports: " + " ".join(["%s/%s" % (port[0], port[1]) for port in source_ports])) self.print_msg(" modules: " + " ".join(modules)) self.print_msg(" destination: " + " ".join(["%s:%s" % (k, v) for k, v in destinations.items()])) self.print_msg(" includes: " + " ".join(sorted(includes))) self.print_msg(" helpers: " + " ".join(sorted(helpers))) def print_icmptype_info(self, icmptype, settings): destinations = settings.getDestinations() description = settings.getDescription() short_description = settings.getShort() if len(destinations) == 0: destinations = [ "ipv4", "ipv6" ] self.print_msg(icmptype) if self.verbose: self.print_msg(" summary: " + short_description) self.print_msg(" description: " + description) self.print_msg(" destination: " + " ".join(destinations)) def print_ipset_info(self, ipset, settings): ipset_type = settings.getType() options = settings.getOptions() entries = settings.getEntries() description = settings.getDescription() short_description = settings.getShort() self.print_msg(ipset) if self.verbose: self.print_msg(" summary: " + short_description) self.print_msg(" description: " + description) self.print_msg(" type: " + ipset_type) self.print_msg(" options: " + " ".join(["%s=%s" % (k, v) if v else k for k, v in options.items()])) self.print_msg(" entries: " + " ".join(entries)) def print_helper_info(self, helper, settings): ports = settings.getPorts() module = settings.getModule() family = settings.getFamily() description = settings.getDescription() short_description = settings.getShort() self.print_msg(helper) if self.verbose: self.print_msg(" summary: " + short_description) self.print_msg(" description: " + description) self.print_msg(" family: " + family) self.print_msg(" module: " + module) self.print_msg(" ports: " + " ".join(["%s/%s" % (port[0], port[1]) for port in ports])) def print_query_result(self, value): if value: self.print_and_exit("yes") else: self.print_and_exit("no", 1) def exception_handler(self, exception_message): if not self.__use_exception_handler: raise self.fail_if_not_authorized(exception_message) code = FirewallError.get_code(str(exception_message)) if code in [ errors.ALREADY_ENABLED, errors.NOT_ENABLED, errors.ZONE_ALREADY_SET, errors.ALREADY_SET ]: self.print_warning("Warning: %s" % exception_message) else: self.print_and_exit("Error: %s" % exception_message, code) def fail_if_not_authorized(self, exception_message): if "NotAuthorizedException" in exception_message: msg = """Authorization failed. Make sure polkit agent is running or run the application as superuser.""" self.print_and_exit(msg, errors.NOT_AUTHORIZED) def deactivate_exception_handler(self): self.__use_exception_handler = False def activate_exception_handler(self): self.__use_exception_handler = True def get_ipset_entries_from_file(self, filename): entries = [ ] entries_set = set() f = open(filename) for line in f: if not line: break line = line.strip() if len(line) < 1 or line[0] in ['#', ';']: continue if line not in entries_set: entries.append(line) entries_set.add(line) f.close() return entries	23.69 KB	2026-03-10 23:36
📁 config	-	2026-03-21 15:45
📁 core	-	2026-03-21 15:45
📄 dbus_utils.py # -- coding: utf-8 -- # # Copyright (C) 2011-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__ = [ "command_of_pid", "pid_of_sender", "uid_of_sender", "user_of_uid", "context_of_sender", "command_of_sender", "user_of_sender", "dbus_to_python", "dbus_signature", "dbus_introspection_prepare_properties", "dbus_introspection_add_properties" ] import dbus import pwd from xml.dom import minidom from firewall.core.logger import log def command_of_pid(pid): """ Get command for pid from /proc """ try: with open("/proc/%d/cmdline" % pid, "r") as f: cmd = f.readlines()[0].replace('\0', " ").strip() except Exception: return None return cmd def pid_of_sender(bus, sender): """ Get pid from sender string using org.freedesktop.DBus.GetConnectionUnixProcessID """ dbus_obj = bus.get_object('org.freedesktop.DBus', '/org/freedesktop/DBus') dbus_iface = dbus.Interface(dbus_obj, 'org.freedesktop.DBus') try: pid = int(dbus_iface.GetConnectionUnixProcessID(sender)) except ValueError: return None return pid def uid_of_sender(bus, sender): """ Get user id from sender string using org.freedesktop.DBus.GetConnectionUnixUser """ dbus_obj = bus.get_object('org.freedesktop.DBus', '/org/freedesktop/DBus') dbus_iface = dbus.Interface(dbus_obj, 'org.freedesktop.DBus') try: uid = int(dbus_iface.GetConnectionUnixUser(sender)) except ValueError: return None return uid def user_of_uid(uid): """ Get user for uid from pwd """ try: pws = pwd.getpwuid(uid) except Exception: return None return pws[0] def context_of_sender(bus, sender): """ Get SELinux context from sender string using org.freedesktop.DBus.GetConnectionSELinuxSecurityContext """ dbus_obj = bus.get_object('org.freedesktop.DBus', '/org/freedesktop/DBus') dbus_iface = dbus.Interface(dbus_obj, 'org.freedesktop.DBus') try: context = dbus_iface.GetConnectionSELinuxSecurityContext(sender) except Exception: return None return "".join(map(chr, dbus_to_python(context))) def command_of_sender(bus, sender): """ Return command of D-Bus sender """ return command_of_pid(pid_of_sender(bus, sender)) def user_of_sender(bus, sender): return user_of_uid(uid_of_sender(bus, sender)) def dbus_to_python(obj, expected_type=None): if obj is None: python_obj = obj elif isinstance(obj, dbus.Boolean): python_obj = bool(obj) elif isinstance(obj, dbus.String): python_obj = str(obj) elif isinstance(obj, dbus.ObjectPath): python_obj = str(obj) elif isinstance(obj, dbus.Byte) or \ isinstance(obj, dbus.Int16) or \ isinstance(obj, dbus.Int32) or \ isinstance(obj, dbus.Int64) or \ isinstance(obj, dbus.UInt16) or \ isinstance(obj, dbus.UInt32) or \ isinstance(obj, dbus.UInt64): python_obj = int(obj) elif isinstance(obj, dbus.Double): python_obj = float(obj) elif isinstance(obj, dbus.Array): python_obj = [dbus_to_python(x) for x in obj] elif isinstance(obj, dbus.Struct): python_obj = tuple([dbus_to_python(x) for x in obj]) elif isinstance(obj, dbus.Dictionary): python_obj = {dbus_to_python(k): dbus_to_python(v) for k, v in obj.items()} elif isinstance(obj, bool) or \ isinstance(obj, str) or isinstance(obj, bytes) or \ isinstance(obj, int) or isinstance(obj, float) or \ isinstance(obj, list) or isinstance(obj, tuple) or \ isinstance(obj, dict): python_obj = obj else: raise TypeError("Unhandled %s" % repr(obj)) if expected_type is not None: if (expected_type == bool and not isinstance(python_obj, bool)) or \ (expected_type == str and not isinstance(python_obj, str)) or \ (expected_type == int and not isinstance(python_obj, int)) or \ (expected_type == float and not isinstance(python_obj, float)) or \ (expected_type == list and not isinstance(python_obj, list)) or \ (expected_type == tuple and not isinstance(python_obj, tuple)) or \ (expected_type == dict and not isinstance(python_obj, dict)): raise TypeError("%s is %s, expected %s" % (python_obj, type(python_obj), expected_type)) return python_obj def dbus_signature(obj): if isinstance(obj, dbus.Boolean): return 'b' elif isinstance(obj, dbus.String): return 's' elif isinstance(obj, dbus.ObjectPath): return 'o' elif isinstance(obj, dbus.Byte): return 'y' elif isinstance(obj, dbus.Int16): return 'n' elif isinstance(obj, dbus.Int32): return 'i' elif isinstance(obj, dbus.Int64): return 'x' elif isinstance(obj, dbus.UInt16): return 'q' elif isinstance(obj, dbus.UInt32): return 'u' elif isinstance(obj, dbus.UInt64): return 't' elif isinstance(obj, dbus.Double): return 'd' elif isinstance(obj, dbus.Array): if len(obj.signature) > 1: return 'a(%s)' % obj.signature else: return 'a%s' % obj.signature elif isinstance(obj, dbus.Struct): return '(%s)' % obj.signature elif isinstance(obj, dbus.Dictionary): return 'a{%s}' % obj.signature else: raise TypeError("Unhandled %s" % repr(obj)) def dbus_introspection_prepare_properties(obj, interface, access=None): if access is None: access = { } if not hasattr(obj, "_fw_dbus_properties"): setattr(obj, "_fw_dbus_properties", { }) dip = getattr(obj, "_fw_dbus_properties") dip[interface] = { } try: _dict = obj.GetAll(interface) except Exception: _dict = { } for key,value in _dict.items(): dip[interface][key] = { "type": dbus_signature(value) } if key in access: dip[interface][key]["access"] = access[key] else: dip[interface][key]["access"] = "read" def dbus_introspection_add_properties(obj, data, interface): doc = minidom.parseString(data) if hasattr(obj, "_fw_dbus_properties"): for node in doc.getElementsByTagName("interface"): if node.hasAttribute("name") and \ node.getAttribute("name") == interface: dip = { } if getattr(obj, "_fw_dbus_properties"): dip = getattr(obj, "_fw_dbus_properties") if interface in dip: for key,value in dip[interface].items(): prop = doc.createElement("property") prop.setAttribute("name", key) prop.setAttribute("type", value["type"]) prop.setAttribute("access", value["access"]) node.appendChild(prop) log.debug10(doc.toxml()) new_data = doc.toxml() doc.unlink() return new_data def dbus_introspection_add_deprecated(obj, data, interface, deprecated_methods, deprecated_signals): doc = minidom.parseString(data) if interface in deprecated_methods: for node in doc.getElementsByTagName("interface"): if node.hasAttribute("name") and \ node.getAttribute("name") == interface: for method_node in node.getElementsByTagName("method"): if method_node.hasAttribute("name") and \ method_node.getAttribute("name") in deprecated_methods[interface]: annotation = doc.createElement("annotation") annotation.setAttribute("name", "org.freedesktop.DBus.Deprecated") annotation.setAttribute("value", "true") method_node.appendChild(annotation) if interface in deprecated_signals: for node in doc.getElementsByTagName("interface"): if node.hasAttribute("name") and \ node.getAttribute("name") == interface: for signal_node in node.getElementsByTagName("signal"): if signal_node.hasAttribute("name") and \ signal_node.getAttribute("name") in deprecated_signals[interface]: annotation = doc.createElement("annotation") annotation.setAttribute("name", "org.freedesktop.DBus.Deprecated") annotation.setAttribute("value", "true") signal_node.appendChild(annotation) log.debug10(doc.toxml()) data = doc.toxml() doc.unlink() return data	9.19 KB	2026-03-10 23:36
📄 errors.py # -- coding: utf-8 -- # # Copyright (C) 2010-2012 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/>. # ALREADY_ENABLED = 11 NOT_ENABLED = 12 COMMAND_FAILED = 13 NO_IPV6_NAT = 14 PANIC_MODE = 15 ZONE_ALREADY_SET = 16 UNKNOWN_INTERFACE = 17 ZONE_CONFLICT = 18 BUILTIN_CHAIN = 19 EBTABLES_NO_REJECT = 20 NOT_OVERLOADABLE = 21 NO_DEFAULTS = 22 BUILTIN_ZONE = 23 BUILTIN_SERVICE = 24 BUILTIN_ICMPTYPE = 25 NAME_CONFLICT = 26 NAME_MISMATCH = 27 PARSE_ERROR = 28 ACCESS_DENIED = 29 UNKNOWN_SOURCE = 30 RT_TO_PERM_FAILED = 31 IPSET_WITH_TIMEOUT = 32 BUILTIN_IPSET = 33 ALREADY_SET = 34 MISSING_IMPORT = 35 DBUS_ERROR = 36 BUILTIN_HELPER = 37 NOT_APPLIED = 38 INVALID_ACTION = 100 INVALID_SERVICE = 101 INVALID_PORT = 102 INVALID_PROTOCOL = 103 INVALID_INTERFACE = 104 INVALID_ADDR = 105 INVALID_FORWARD = 106 INVALID_ICMPTYPE = 107 INVALID_TABLE = 108 INVALID_CHAIN = 109 INVALID_TARGET = 110 INVALID_IPV = 111 INVALID_ZONE = 112 INVALID_PROPERTY = 113 INVALID_VALUE = 114 INVALID_OBJECT = 115 INVALID_NAME = 116 INVALID_FILENAME = 117 INVALID_DIRECTORY = 118 INVALID_TYPE = 119 INVALID_SETTING = 120 INVALID_DESTINATION = 121 INVALID_RULE = 122 INVALID_LIMIT = 123 INVALID_FAMILY = 124 INVALID_LOG_LEVEL = 125 INVALID_AUDIT_TYPE = 126 INVALID_MARK = 127 INVALID_CONTEXT = 128 INVALID_COMMAND = 129 INVALID_USER = 130 INVALID_UID = 131 INVALID_MODULE = 132 INVALID_PASSTHROUGH = 133 INVALID_MAC = 134 INVALID_IPSET = 135 INVALID_ENTRY = 136 INVALID_OPTION = 137 INVALID_HELPER = 138 INVALID_PRIORITY = 139 INVALID_POLICY = 140 INVALID_LOG_PREFIX = 141 INVALID_NFLOG_GROUP = 142 INVALID_NFLOG_QUEUE = 143 MISSING_TABLE = 200 MISSING_CHAIN = 201 MISSING_PORT = 202 MISSING_PROTOCOL = 203 MISSING_ADDR = 204 MISSING_NAME = 205 MISSING_SETTING = 206 MISSING_FAMILY = 207 RUNNING_BUT_FAILED = 251 NOT_RUNNING = 252 NOT_AUTHORIZED = 253 UNKNOWN_ERROR = 254 import sys class FirewallError(Exception): def __init__(self, code, msg=None): self.code = code self.msg = msg def __repr__(self): return '%s(%r, %r)' % (self.__class__, self.code, self.msg) def __str__(self): if self.msg: return "%s: %s" % (self.errors[self.code], self.msg) return self.errors[self.code] def get_code(msg): if msg is not None and ":" in msg: idx = msg.index(":") ecode = msg[:idx] else: ecode = msg try: code = FirewallError.codes[ecode] except KeyError: code = UNKNOWN_ERROR return code get_code = staticmethod(get_code) mod = sys.modules[FirewallError.__module__] FirewallError.errors = { getattr(mod,varname) : varname for varname in dir(mod) if not varname.startswith("_") and \ type(getattr(mod,varname)) == int } FirewallError.codes = { FirewallError.errors[code] : code for code in FirewallError.errors }	4.04 KB	2026-03-10 23:36
📄 functions.py # -- coding: utf-8 -- # # Copyright (C) 2007,2008,2011,2012 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__ = [ "getPortID", "getPortRange", "portStr", "getServiceName", "checkIP", "checkIP6", "checkIPnMask", "checkIP6nMask", "checkProtocol", "checkInterface", "checkUINT16", "checkUINT32", "firewalld_is_active", "tempFile", "readfile", "writefile", "enable_ip_forwarding", "check_port", "check_address", "check_single_address", "check_mac", "uniqify", "ppid_of_pid", "max_zone_name_len", "checkUser", "checkUid", "checkCommand", "checkContext", "joinArgs", "splitArgs", "max_policy_name_len", "checkTcpMssClamp", "stripNonPrintableCharacters"] import socket import os import os.path import shlex import pipes import string import tempfile from firewall.core.logger import log from firewall.config import FIREWALLD_TEMPDIR, FIREWALLD_PIDFILE NOPRINT_TRANS_TABLE = { # Limit to C0 and C1 code points. Building entries for all unicode code # points requires too much memory. # C0 = [0, 31] # C1 = [127, 159] # i: None for i in range(0, 160) if not (i > 31 and i < 127) } def getPortID(port): """ Check and Get port id from port string or port id using socket.getservbyname @param port port string or port id @return Port id if valid, -1 if port can not be found and -2 if port is too big """ if isinstance(port, int): _id = port else: if port: port = port.strip() try: _id = int(port) except ValueError: try: _id = socket.getservbyname(port) except socket.error: return -1 if _id > 65535: return -2 return _id def getPortRange(ports): """ Get port range for port range string or single port id @param ports an integer or port string or port range string @return Array containing start and end port id for a valid range or -1 if port can not be found and -2 if port is too big for integer input or -1 for invalid ranges or None if the range is ambiguous. """ # (port, port) or [port, port] case if isinstance(ports, tuple) or isinstance(ports, list): return ports # "<port-id>" case if isinstance(ports, int) or ports.isdigit(): id1 = getPortID(ports) if id1 >= 0: return (id1,) return id1 splits = ports.split("-") # "<port-id>-<port-id>" case if len(splits) == 2 and splits[0].isdigit() and splits[1].isdigit(): id1 = getPortID(splits[0]) id2 = getPortID(splits[1]) if id1 >= 0 and id2 >= 0: if id1 < id2: return (id1, id2) elif id1 > id2: return (id2, id1) else: # ids are the same return (id1,) # everything else "<port-str>[-<port-str>]" matched = [ ] for i in range(len(splits), 0, -1): id1 = getPortID("-".join(splits[:i])) port2 = "-".join(splits[i:]) if len(port2) > 0: id2 = getPortID(port2) if id1 >= 0 and id2 >= 0: if id1 < id2: matched.append((id1, id2)) elif id1 > id2: matched.append((id2, id1)) else: matched.append((id1, )) else: if id1 >= 0: matched.append((id1,)) if i == len(splits): # full match, stop here break if len(matched) < 1: return -1 elif len(matched) > 1: return None return matched[0] def portStr(port, delimiter=":"): """ Create port and port range string @param port port or port range int or [int, int] @param delimiter of the output string for port ranges, default ':' @return Port or port range string, empty string if port isn't specified, None if port or port range is not valid """ if port == "": return "" _range = getPortRange(port) if isinstance(_range, int) and _range < 0: return None elif len(_range) == 1: return "%s" % _range else: return "%s%s%s" % (_range[0], delimiter, _range[1]) def portInPortRange(port, range): _port = getPortRange(port) _range = getPortRange(range) if len(_port) == 1: if len(_range) == 1: return getPortID(_port[0]) == getPortID(_range[0]) if len(_range) == 2 and \ getPortID(_port[0]) >= getPortID(_range[0]) and getPortID(_port[0]) <= getPortID(_range[1]): return True elif len(_port) == 2: if len(_range) == 2 and \ getPortID(_port[0]) >= getPortID(_range[0]) and getPortID(_port[0]) <= getPortID(_range[1]) and \ getPortID(_port[1]) >= getPortID(_range[0]) and getPortID(_port[1]) <= getPortID(_range[1]): return True return False def coalescePortRange(new_range, ranges): """ Coalesce a port range with existing list of port ranges @param new_range tuple/list/string @param ranges list of tuple/list/string @return tuple of (list of ranges added after coalescing, list of removed original ranges) """ coalesced_range = getPortRange(new_range) # normalize singleton ranges, e.g. (x,) --> (x,x) if len(coalesced_range) == 1: coalesced_range = (coalesced_range[0], coalesced_range[0]) _ranges = map(getPortRange, ranges) _ranges = sorted(map(lambda x: (x[0],x[0]) if len(x) == 1 else x, _ranges), key=lambda x: x[0]) removed_ranges = [] for range in _ranges: if coalesced_range[0] <= range[0] and coalesced_range[1] >= range[1]: # new range covers this removed_ranges.append(range) elif coalesced_range[0] <= range[0] and coalesced_range[1] < range[1] and \ coalesced_range[1] >= range[0]: # expand beginning of range removed_ranges.append(range) coalesced_range = (coalesced_range[0], range[1]) elif coalesced_range[0] > range[0] and coalesced_range[1] >= range[1] and \ coalesced_range[0] <= range[1]: # expand end of range removed_ranges.append(range) coalesced_range = (range[0], coalesced_range[1]) # normalize singleton ranges, e.g. (x,x) --> (x,) removed_ranges = list(map(lambda x: (x[0],) if x[0] == x[1] else x, removed_ranges)) if coalesced_range[0] == coalesced_range[1]: coalesced_range = (coalesced_range[0],) return ([coalesced_range], removed_ranges) def breakPortRange(remove_range, ranges): """ break a port range from existing list of port ranges @param remove_range tuple/list/string @param ranges list of tuple/list/string @return tuple of (list of ranges added after breaking up, list of removed original ranges) """ remove_range = getPortRange(remove_range) # normalize singleton ranges, e.g. (x,) --> (x,x) if len(remove_range) == 1: remove_range = (remove_range[0], remove_range[0]) _ranges = map(getPortRange, ranges) _ranges = sorted(map(lambda x: (x[0],x[0]) if len(x) == 1 else x, _ranges), key=lambda x: x[0]) removed_ranges = [] added_ranges = [] for range in _ranges: if remove_range[0] <= range[0] and remove_range[1] >= range[1]: # remove entire range removed_ranges.append(range) elif remove_range[0] <= range[0] and remove_range[1] < range[1] and \ remove_range[1] >= range[0]: # remove from beginning of range removed_ranges.append(range) added_ranges.append((remove_range[1] + 1, range[1])) elif remove_range[0] > range[0] and remove_range[1] >= range[1] and \ remove_range[0] <= range[1]: # remove from end of range removed_ranges.append(range) added_ranges.append((range[0], remove_range[0] - 1)) elif remove_range[0] > range[0] and remove_range[1] < range[1]: # remove inside range removed_ranges.append(range) added_ranges.append((range[0], remove_range[0] - 1)) added_ranges.append((remove_range[1] + 1, range[1])) # normalize singleton ranges, e.g. (x,x) --> (x,) removed_ranges = list(map(lambda x: (x[0],) if x[0] == x[1] else x, removed_ranges)) added_ranges = list(map(lambda x: (x[0],) if x[0] == x[1] else x, added_ranges)) return (added_ranges, removed_ranges) def getServiceName(port, proto): """ Check and Get service name from port and proto string combination using socket.getservbyport @param port string or id @param protocol string @return Service name if port and protocol are valid, else None """ try: name = socket.getservbyport(int(port), proto) except socket.error: return None return name def checkIP(ip): """ Check IPv4 address. @param ip address string @return True if address is valid, else False """ try: socket.inet_pton(socket.AF_INET, ip) except socket.error: return False return True def normalizeIP6(ip): """ Normalize the IPv6 address This is mostly about converting URL-like IPv6 address to normal ones. e.g. [1234::4321] --> 1234:4321 """ return ip.strip("[]") def checkIP6(ip): """ Check IPv6 address. @param ip address string @return True if address is valid, else False """ try: socket.inet_pton(socket.AF_INET6, normalizeIP6(ip)) except socket.error: return False return True def checkIPnMask(ip): index = ip.find("/") if index != -1: addr = ip[:index] mask = ip[index + 1:] if not addr or not mask: return False else: addr = ip mask = None if not checkIP(addr): return False if mask: if "." in mask: return checkIP(mask) else: try: i = int(mask) except ValueError: return False if i < 0 or i > 32: return False return True def stripNonPrintableCharacters(rule_str): return rule_str.translate(NOPRINT_TRANS_TABLE) def checkIP6nMask(ip): index = ip.find("/") if index != -1: addr = ip[:index] mask = ip[index + 1:] if not addr or not mask: return False else: addr = ip mask = None if not checkIP6(addr): return False if mask: try: i = int(mask) except ValueError: return False if i < 0 or i > 128: return False return True def checkProtocol(protocol): try: i = int(protocol) except ValueError: # string try: socket.getprotobyname(protocol) except socket.error: return False else: if i < 0 or i > 255: return False return True def checkTcpMssClamp(tcp_mss_clamp_value): if tcp_mss_clamp_value: if tcp_mss_clamp_value.isdigit(): if int(tcp_mss_clamp_value) < 536: return False elif tcp_mss_clamp_value == "None": return True elif tcp_mss_clamp_value != "pmtu": return False return True def checkInterface(iface): """ Check interface string @param interface string @return True if interface is valid (maximum 16 chars and does not contain ' ', '/', '!', ':', ''), else False """ if not iface or len(iface) > 16: return False for ch in [ ' ', '/', '!', '' ]: # !:* are limits for iptables <= 1.4.5 if ch in iface: return False # disabled old iptables check #if iface == "+": # # limit for iptables <= 1.4.5 # return False return True def checkUINT16(val): try: x = int(val, 0) except ValueError: return False else: if x >= 0 and x <= 65535: return True return False def checkUINT32(val): try: x = int(val, 0) except ValueError: return False else: if x >= 0 and x <= 4294967295: return True return False def firewalld_is_active(): """ Check if firewalld is active @return True if there is a firewalld pid file and the pid is used by firewalld """ if not os.path.exists(FIREWALLD_PIDFILE): return False try: with open(FIREWALLD_PIDFILE, "r") as fd: pid = fd.readline() except Exception: return False if not os.path.exists("/proc/%s" % pid): return False try: with open("/proc/%s/cmdline" % pid, "r") as fd: cmdline = fd.readline() except Exception: return False if "firewalld" in cmdline: return True return False def tempFile(): try: if not os.path.exists(FIREWALLD_TEMPDIR): os.mkdir(FIREWALLD_TEMPDIR, 0o750) return tempfile.NamedTemporaryFile(mode='wt', prefix="temp.", dir=FIREWALLD_TEMPDIR, delete=False) except Exception as msg: log.error("Failed to create temporary file: %s" % msg) raise def readfile(filename): try: with open(filename, "r") as f: return f.readlines() except Exception as e: log.error('Failed to read file "%s": %s' % (filename, e)) return None def writefile(filename, line): try: with open(filename, "w") as f: f.write(line) except Exception as e: log.error('Failed to write to file "%s": %s' % (filename, e)) return False return True def enable_ip_forwarding(ipv): if ipv == "ipv4": return writefile("/proc/sys/net/ipv4/ip_forward", "1\n") elif ipv == "ipv6": return writefile("/proc/sys/net/ipv6/conf/all/forwarding", "1\n") return False def get_nf_conntrack_short_name(module): return module.replace("_","-").replace("nf-conntrack-", "") def check_port(port): _range = getPortRange(port) if _range == -2 or _range == -1 or _range is None or \ (len(_range) == 2 and _range[0] >= _range[1]): if _range == -2: log.debug2("'%s': port > 65535" % port) elif _range == -1: log.debug2("'%s': port is invalid" % port) elif _range is None: log.debug2("'%s': port is ambiguous" % port) elif len(_range) == 2 and _range[0] >= _range[1]: log.debug2("'%s': range start >= end" % port) return False return True def check_address(ipv, source): if ipv == "ipv4": return checkIPnMask(source) elif ipv == "ipv6": return checkIP6nMask(source) else: return False def check_single_address(ipv, source): if ipv == "ipv4": return checkIP(source) elif ipv == "ipv6": return checkIP6(source) else: return False def check_mac(mac): if len(mac) == 12+5: # 0 1 : 3 4 : 6 7 : 9 10 : 12 13 : 15 16 for i in (2, 5, 8, 11, 14): if mac[i] != ":": return False for i in (0, 1, 3, 4, 6, 7, 9, 10, 12, 13, 15, 16): if mac[i] not in string.hexdigits: return False return True return False def uniqify(_list): # removes duplicates from list, whilst preserving order output = [] for x in _list: if x not in output: output.append(x) return output def ppid_of_pid(pid): """ Get parent for pid """ try: f = os.popen("ps -o ppid -h -p %d 2>/dev/null" % pid) pid = int(f.readlines()[0].strip()) f.close() except Exception: return None return pid def max_policy_name_len(): """ iptables limits length of chain to (currently) 28 chars. The longest chain we create is POST_<policy>_allow, which leaves 28 - 11 = 17 chars for <policy>. """ from firewall.core.ipXtables import POLICY_CHAIN_PREFIX from firewall.core.base import SHORTCUTS longest_shortcut = max(map(len, SHORTCUTS.values())) return 28 - (longest_shortcut + len(POLICY_CHAIN_PREFIX) + len("_allow")) def max_zone_name_len(): """ Netfilter limits length of chain to (currently) 28 chars. The longest chain we create is POST_<zone>_allow, which leaves 28 - 11 = 17 chars for <zone>. """ from firewall.core.base import SHORTCUTS longest_shortcut = max(map(len, SHORTCUTS.values())) return 28 - (longest_shortcut + len("__allow")) def checkUser(user): if len(user) < 1 or len(user) > os.sysconf('SC_LOGIN_NAME_MAX'): return False for c in user: if c not in string.ascii_letters and \ c not in string.digits and \ c not in [ ".", "-", "_", "$" ]: return False return True def checkUid(uid): if isinstance(uid, str): try: uid = int(uid) except ValueError: return False if uid >= 0 and uid <= 2**31-1: return True return False def checkCommand(command): if len(command) < 1 or len(command) > 1024: return False for ch in [ "\|", "\n", "\0" ]: if ch in command: return False if command[0] != "/": return False return True def checkContext(context): splits = context.split(":") if len(splits) not in [4, 5]: return False # user ends with _u if not root if splits[0] != "root" and splits[0][-2:] != "_u": return False # role ends with _r if splits[1][-2:] != "_r": return False # type ends with _t if splits[2][-2:] != "_t": return False # level might also contain : if len(splits[3]) < 1: return False return True def joinArgs(args): if "quote" in dir(shlex): return " ".join(shlex.quote(a) for a in args) else: return " ".join(pipes.quote(a) for a in args) def splitArgs(_string): return shlex.split(_string)	18.54 KB	2026-03-10 23:36
📄 fw_types.py # -- coding: utf-8 -- # # Copyright (C) 2013-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__ = [ "LastUpdatedOrderedDict" ] class LastUpdatedOrderedDict(object): def __init__(self, x=None): self._dict = { } self._list = [ ] if x: self.update(x) def clear(self): del self._list[:] self._dict.clear() def update(self, x): for key,value in x.items(): self[key] = value def items(self): return [(key, self[key]) for key in self._list] def __delitem__(self, key): if key in self._dict: self._list.remove(key) del self._dict[key] def __repr__(self): return '%s([%s])' % (self.__class__.__name__, ', '.join( ['(%r, %r)' % (key, self[key]) for key in self._list])) def __setitem__(self, key, value): if key not in self._dict: self._list.append(key) self._dict[key] = value def __getitem__(self, key): if type(key) == int: return self._list[key] else: return self._dict[key] def __len__(self): return len(self._list) def copy(self): return LastUpdatedOrderedDict(self) def keys(self): return self._list[:] def values(self): return [ self[key] for key in self._list ] def setdefault(self, key, value=None): if key in self: return self[key] else: self[key] = value return value	2.14 KB	2026-03-10 23:36
📁 server	-	2026-03-21 15:45

Terminal