Wala shell

Advanced PHP File Manager

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

Name	Size	Last Modified	Actions
📁 ..	-	2026-03-21 15:45
📄 __init__.py # -- coding: utf-8 -- # # Copyright (C) 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/>. # # fix xmlplus to be compatible with the python xml sax parser and python 3 # by adding __contains__ to xml.sax.xmlreader.AttributesImpl import xml if "_xmlplus" in xml.__file__: from xml.sax.xmlreader import AttributesImpl if not hasattr(AttributesImpl, "__contains__"): # this is missing: def __AttributesImpl__contains__(self, name): return name in getattr(self, "_attrs") # add it using the name __contains__ setattr(AttributesImpl, "__contains__", __AttributesImpl__contains__) from xml.sax.saxutils import XMLGenerator if not hasattr(XMLGenerator, "_write"): # this is missing: def __XMLGenerator_write(self, text): getattr(self, "_out").write(text) # add it using the name _write setattr(XMLGenerator, "_write", __XMLGenerator_write)	1.56 KB	2026-03-10 23:36
📁 __pycache__	-	2026-03-21 15:45
📄 direct.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/>. # import xml.sax as sax import os import io import shutil from firewall import config from firewall.fw_types import LastUpdatedOrderedDict from firewall.functions import splitArgs, joinArgs from firewall.core.io.io_object import IO_Object, IO_Object_ContentHandler, \ IO_Object_XMLGenerator from firewall.core.logger import log from firewall.core import ipXtables from firewall.core import ebtables from firewall import errors from firewall.errors import FirewallError class direct_ContentHandler(IO_Object_ContentHandler): def __init__(self, item): IO_Object_ContentHandler.__init__(self, item) self.direct = False def startElement(self, name, attrs): IO_Object_ContentHandler.startElement(self, name, attrs) self.item.parser_check_element_attrs(name, attrs) if name == "direct": if self.direct: raise FirewallError(errors.PARSE_ERROR, "More than one direct tag.") self.direct = True elif name == "chain": if not self.direct: log.error("Parse Error: chain outside of direct") return ipv = attrs["ipv"] table = attrs["table"] chain = attrs["chain"] self.item.add_chain(ipv, table, chain) elif name == "rule": if not self.direct: log.error("Parse Error: rule outside of direct") return ipv = attrs["ipv"] if ipv not in [ "ipv4", "ipv6", "eb" ]: raise FirewallError(errors.INVALID_IPV, "'%s' not from {'ipv4'\|'ipv6'\|'eb'}" % ipv) table = attrs["table"] chain = attrs["chain"] try: priority = int(attrs["priority"]) except ValueError: log.error("Parse Error: %s is not a valid priority" % attrs["priority"]) return self._rule = [ ipv, table, chain, priority ] elif name == "passthrough": if not self.direct: log.error("Parse Error: command outside of direct") return ipv = attrs["ipv"] self._passthrough = [ ipv ] else: log.error('Unknown XML element %s' % name) return def endElement(self, name): IO_Object_ContentHandler.endElement(self, name) if name == "rule": if self._element: # add arguments self._rule.append(splitArgs(self._element)) self.item.add_rule(self._rule) else: log.error("Error: rule does not have any arguments, ignoring.") self._rule = None elif name == "passthrough": if self._element: # add arguments self._passthrough.append(splitArgs(self._element)) self.item.add_passthrough(self._passthrough) else: log.error("Error: passthrough does not have any arguments, " + "ignoring.") self._passthrough = None class Direct(IO_Object): """ Direct class """ IMPORT_EXPORT_STRUCTURE = ( # chain: [ ipv, table, [ chain ] ] ( "chains", [ ( "", "", "" ), ], ), # a(sss) # rule: [ ipv, table, chain, [ priority, [ arg ] ] ] ( "rules", [ ( "", "", "", 0, [ "" ] ), ], ), # a(sssias) # passthrough: [ ipv, [ [ arg ] ] ] ( "passthroughs", [ ( "", [ "" ]), ], ), # a(sas) ) DBUS_SIGNATURE = '(a(sss)a(sssias)a(sas))' PARSER_REQUIRED_ELEMENT_ATTRS = { "direct": None, "chain": [ "ipv", "table", "chain" ], "rule": [ "ipv", "table", "chain", "priority" ], "passthrough": [ "ipv" ] } PARSER_OPTIONAL_ELEMENT_ATTRS = { } def __init__(self, filename): super(Direct, self).__init__() self.filename = filename self.chains = LastUpdatedOrderedDict() self.rules = LastUpdatedOrderedDict() self.passthroughs = LastUpdatedOrderedDict() def _check_config(self, conf, item, all_conf, all_io_objects): pass # check arg lists def export_config(self): ret = [ ] x = [ ] for key in self.chains: for chain in self.chains[key]: x.append(tuple(list(key) + list([chain]))) ret.append(x) x = [ ] for key in self.rules: for rule in self.rules[key]: x.append(tuple((key[0], key[1], key[2], rule[0], list(rule[1])))) ret.append(x) x = [ ] for key in self.passthroughs: for rule in self.passthroughs[key]: x.append(tuple((key, list(rule)))) ret.append(x) return tuple(ret) def import_config(self, conf, all_io_objects): self.cleanup() self.check_config(conf) for i,(element,dummy) in enumerate(self.IMPORT_EXPORT_STRUCTURE): if element == "chains": for x in conf[i]: self.add_chain(x) if element == "rules": for x in conf[i]: self.add_rule(x) if element == "passthroughs": for x in conf[i]: self.add_passthrough(*x) def cleanup(self): self.chains.clear() self.rules.clear() self.passthroughs.clear() def output(self): print("chains") for key in self.chains: print(" (%s, %s): %s" % (key[0], key[1], ",".join(self.chains[key]))) print("rules") for key in self.rules: print(" (%s, %s, %s):" % (key[0], key[1], key[2])) for (priority,args) in self.rules[key]: print(" (%d, ('%s'))" % (priority, "','".join(args))) print("passthroughs") for key in self.passthroughs: print(" %s:" % (key)) for args in self.passthroughs[key]: print(" ('%s')" % ("','".join(args))) def _check_ipv(self, ipv): ipvs = ['ipv4', 'ipv6', 'eb'] if ipv not in ipvs: raise FirewallError(errors.INVALID_IPV, "'%s' not in '%s'" % (ipv, ipvs)) def _check_ipv_table(self, ipv, table): self._check_ipv(ipv) tables = ipXtables.BUILT_IN_CHAINS.keys() if ipv in ['ipv4', 'ipv6'] \ else ebtables.BUILT_IN_CHAINS.keys() if table not in tables: raise FirewallError(errors.INVALID_TABLE, "'%s' not in '%s'" % (table, tables)) # chains def add_chain(self, ipv, table, chain): self._check_ipv_table(ipv, table) key = (ipv, table) if key not in self.chains: self.chains[key] = [ ] if chain not in self.chains[key]: self.chains[key].append(chain) else: log.warning("Chain '%s' for table '%s' with ipv '%s' " % \ (chain, table, ipv) + "already in list, ignoring") def remove_chain(self, ipv, table, chain): self._check_ipv_table(ipv, table) key = (ipv, table) if key in self.chains and chain in self.chains[key]: self.chains[key].remove(chain) if len(self.chains[key]) == 0: del self.chains[key] else: raise ValueError( \ "Chain '%s' with table '%s' with ipv '%s' not in list" % \ (chain, table, ipv)) def query_chain(self, ipv, table, chain): self._check_ipv_table(ipv, table) key = (ipv, table) return (key in self.chains and chain in self.chains[key]) def get_chains(self, ipv, table): self._check_ipv_table(ipv, table) key = (ipv, table) if key in self.chains: return self.chains[key] else: raise ValueError("No chains for table '%s' with ipv '%s'" % \ (table, ipv)) def get_all_chains(self): return self.chains # rules def add_rule(self, ipv, table, chain, priority, args): self._check_ipv_table(ipv, table) key = (ipv, table, chain) if key not in self.rules: self.rules[key] = LastUpdatedOrderedDict() value = (priority, tuple(args)) if value not in self.rules[key]: self.rules[key][value] = priority else: log.warning("Rule '%s' for table '%s' and chain '%s' " % \ ("',".join(args), table, chain) + "with ipv '%s' and priority %d " % (ipv, priority) + "already in list, ignoring") def remove_rule(self, ipv, table, chain, priority, args): self._check_ipv_table(ipv, table) key = (ipv, table, chain) value = (priority, tuple(args)) if key in self.rules and value in self.rules[key]: del self.rules[key][value] if len(self.rules[key]) == 0: del self.rules[key] else: raise ValueError("Rule '%s' for table '%s' and chain '%s' " % \ ("',".join(args), table, chain) + \ "with ipv '%s' and priority %d not in list" % (ipv, priority)) def remove_rules(self, ipv, table, chain): self._check_ipv_table(ipv, table) key = (ipv, table, chain) if key in self.rules: for value in self.rules[key].keys(): del self.rules[key][value] if len(self.rules[key]) == 0: del self.rules[key] def query_rule(self, ipv, table, chain, priority, args): self._check_ipv_table(ipv, table) key = (ipv, table, chain) value = (priority, tuple(args)) return (key in self.rules and value in self.rules[key]) def get_rules(self, ipv, table, chain): self._check_ipv_table(ipv, table) key = (ipv, table, chain) if key in self.rules: return self.rules[key] else: raise ValueError("No rules for table '%s' and chain '%s' " %\ (table, chain) + "with ipv '%s'" % (ipv)) def get_all_rules(self): return self.rules # # passthrough # def add_passthrough(self, ipv, args): self._check_ipv(ipv) if ipv not in self.passthroughs: self.passthroughs[ipv] = [ ] if args not in self.passthroughs[ipv]: self.passthroughs[ipv].append(args) else: log.warning("Passthrough '%s' for ipv '%s'" % \ ("',".join(args), ipv) + "already in list, ignoring") def remove_passthrough(self, ipv, args): self._check_ipv(ipv) if ipv in self.passthroughs and args in self.passthroughs[ipv]: self.passthroughs[ipv].remove(args) if len(self.passthroughs[ipv]) == 0: del self.passthroughs[ipv] else: raise ValueError("Passthrough '%s' for ipv '%s'" % \ ("',".join(args), ipv) + "not in list") def query_passthrough(self, ipv, args): self._check_ipv(ipv) return ipv in self.passthroughs and args in self.passthroughs[ipv] def get_passthroughs(self, ipv): self._check_ipv(ipv) if ipv in self.passthroughs: return self.passthroughs[ipv] else: raise ValueError("No passthroughs for ipv '%s'" % (ipv)) def get_all_passthroughs(self): return self.passthroughs # read def read(self): self.cleanup() if not self.filename.endswith(".xml"): raise FirewallError(errors.INVALID_NAME, "'%s' is missing .xml suffix" % self.filename) handler = direct_ContentHandler(self) parser = sax.make_parser() parser.setContentHandler(handler) with open(self.filename, "rb") as f: source = sax.InputSource(None) source.setByteStream(f) try: parser.parse(source) except sax.SAXParseException as msg: raise FirewallError(errors.INVALID_TYPE, "Not a valid file: %s" % \ msg.getException()) def write(self): if os.path.exists(self.filename): try: shutil.copy2(self.filename, "%s.old" % self.filename) except Exception as msg: raise IOError("Backup of '%s' failed: %s" % (self.filename, msg)) if not os.path.exists(config.ETC_FIREWALLD): os.mkdir(config.ETC_FIREWALLD, 0o750) f = io.open(self.filename, mode='wt', encoding='UTF-8') handler = IO_Object_XMLGenerator(f) handler.startDocument() # start whitelist element handler.startElement("direct", { }) handler.ignorableWhitespace("\n") # chains for key in self.chains: (ipv, table) = key for chain in self.chains[key]: handler.ignorableWhitespace(" ") handler.simpleElement("chain", { "ipv": ipv, "table": table, "chain": chain }) handler.ignorableWhitespace("\n") # rules for key in self.rules: (ipv, table, chain) = key for (priority, args) in self.rules[key]: if len(args) < 1: continue handler.ignorableWhitespace(" ") handler.startElement("rule", { "ipv": ipv, "table": table, "chain": chain, "priority": "%d" % priority }) handler.ignorableWhitespace(sax.saxutils.escape(joinArgs(args))) handler.endElement("rule") handler.ignorableWhitespace("\n") # passthroughs for ipv in self.passthroughs: for args in self.passthroughs[ipv]: if len(args) < 1: continue handler.ignorableWhitespace(" ") handler.startElement("passthrough", { "ipv": ipv }) handler.ignorableWhitespace(sax.saxutils.escape(joinArgs(args))) handler.endElement("passthrough") handler.ignorableWhitespace("\n") # end zone element handler.endElement("direct") handler.ignorableWhitespace("\n") handler.endDocument() f.close() del handler	15.21 KB	2026-03-10 23:36
📄 firewalld_conf.py # -- coding: utf-8 -- # # Copyright (C) 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/>. # import os.path import io import tempfile import shutil from firewall import config from firewall.core.logger import log from firewall import errors valid_keys = [ "DefaultZone", "MinimalMark", "CleanupOnExit", "CleanupModulesOnExit", "Lockdown", "IPv6_rpfilter", "IndividualCalls", "LogDenied", "AutomaticHelpers", "FirewallBackend", "FlushAllOnReload", "RFC3964_IPv4", "AllowZoneDrifting", "ReloadPolicy", "NftablesTableOwner" ] class firewalld_conf(object): def __init__(self, filename): self._config = { } self._deleted = [ ] self.filename = filename self.clear() def clear(self): self._config = { } self._deleted = [ ] def cleanup(self): self._config.clear() self._deleted = [ ] def get(self, key): return self._config.get(key.strip()) def set(self, key, value): _key = key.strip() self._config[_key] = value.strip() if _key in self._deleted: self._deleted.remove(_key) def __str__(self): s = "" for (key,value) in self._config.items(): if s: s += '\n' s += '%s=%s' % (key, value) return s def set_defaults(self): self.set("DefaultZone", config.FALLBACK_ZONE) self.set("MinimalMark", str(config.FALLBACK_MINIMAL_MARK)) self.set("CleanupOnExit", "yes" if config.FALLBACK_CLEANUP_ON_EXIT else "no") self.set("CleanupModulesOnExit", "yes" if config.FALLBACK_CLEANUP_MODULES_ON_EXIT else "no") self.set("Lockdown", "yes" if config.FALLBACK_LOCKDOWN else "no") self.set("IPv6_rpfilter", config.FALLBACK_IPV6_RPFILTER) self.set("IndividualCalls", "yes" if config.FALLBACK_INDIVIDUAL_CALLS else "no") self.set("LogDenied", config.FALLBACK_LOG_DENIED) self.set("AutomaticHelpers", config.FALLBACK_AUTOMATIC_HELPERS) self.set("FirewallBackend", config.FALLBACK_FIREWALL_BACKEND) self.set("FlushAllOnReload", "yes" if config.FALLBACK_FLUSH_ALL_ON_RELOAD else "no") self.set("ReloadPolicy", config.FALLBACK_RELOAD_POLICY) self.set("RFC3964_IPv4", "yes" if config.FALLBACK_RFC3964_IPV4 else "no") self.set("AllowZoneDrifting", "yes" if config.FALLBACK_ALLOW_ZONE_DRIFTING else "no") self.set( "NftablesTableOwner", "yes" if config.FALLBACK_NFTABLES_TABLE_OWNER else "no", ) def sanity_check(self): if self.get("FirewallBackend") == "iptables" and self.get("IPv6_rpfilter") in ( "loose-forward", "strict-forward", ): raise errors.FirewallError( errors.INVALID_VALUE, f"IPv6_rpfilter={self.get('IPv6_rpfilter')} is incompatible " "with FirewallBackend=iptables. This is a limitation " "of the iptables backend.", ) # load self.filename def read(self): self.clear() try: f = open(self.filename, "r") except Exception as msg: log.error("Failed to load '%s': %s", self.filename, msg) self.set_defaults() raise for line in f: if not line: break line = line.strip() if len(line) < 1 or line[0] in ['#', ';']: continue # get key/value pair pair = [ x.strip() for x in line.split("=") ] if len(pair) != 2: log.error("Invalid option definition: '%s'", line.strip()) continue elif pair[0] not in valid_keys: log.error("Invalid option: '%s'", line.strip()) continue elif pair[1] == '': log.error("Missing value: '%s'", line.strip()) continue elif self._config.get(pair[0]) is not None: log.error("Duplicate option definition: '%s'", line.strip()) continue self._config[pair[0]] = pair[1] f.close() # check default zone if not self.get("DefaultZone"): log.error("DefaultZone is not set, using default value '%s'", config.FALLBACK_ZONE) self.set("DefaultZone", str(config.FALLBACK_ZONE)) # check minimal mark value = self.get("MinimalMark") try: int(value) except (ValueError, TypeError): if value is not None: log.warning("MinimalMark '%s' is not valid, using default " "value '%d'", value if value else '', config.FALLBACK_MINIMAL_MARK) self.set("MinimalMark", str(config.FALLBACK_MINIMAL_MARK)) # check cleanup on exit value = self.get("CleanupOnExit") if not value or value.lower() not in [ "no", "false", "yes", "true" ]: if value is not None: log.warning("CleanupOnExit '%s' is not valid, using default " "value %s", value if value else '', config.FALLBACK_CLEANUP_ON_EXIT) self.set("CleanupOnExit", "yes" if config.FALLBACK_CLEANUP_ON_EXIT else "no") # check module cleanup on exit value = self.get("CleanupModulesOnExit") if not value or value.lower() not in [ "no", "false", "yes", "true" ]: if value is not None: log.warning("CleanupModulesOnExit '%s' is not valid, using default " "value %s", value if value else '', config.FALLBACK_CLEANUP_MODULES_ON_EXIT) self.set("CleanupModulesOnExit", "yes" if config.FALLBACK_CLEANUP_MODULES_ON_EXIT else "no") # check lockdown value = self.get("Lockdown") if not value or value.lower() not in [ "yes", "true", "no", "false" ]: if value is not None: log.warning("Lockdown '%s' is not valid, using default " "value %s", value if value else '', config.FALLBACK_LOCKDOWN) self.set("Lockdown", "yes" if config.FALLBACK_LOCKDOWN else "no") # check ipv6_rpfilter value = self.get("IPv6_rpfilter") if not value or value.lower() not in config.IPV6_RPFILTER_VALUES: if value is not None: log.warning( "IPv6_rpfilter '%s' is not valid, using default " "value %s", value if value else "", config.FALLBACK_IPV6_RPFILTER, ) self.set("IPv6_rpfilter", config.FALLBACK_IPV6_RPFILTER) # check individual calls value = self.get("IndividualCalls") if not value or value.lower() not in [ "yes", "true", "no", "false" ]: if value is not None: log.warning("IndividualCalls '%s' is not valid, using default " "value %s", value if value else '', config.FALLBACK_INDIVIDUAL_CALLS) self.set("IndividualCalls", "yes" if config.FALLBACK_INDIVIDUAL_CALLS else "no") # check log denied value = self.get("LogDenied") if not value or value not in config.LOG_DENIED_VALUES: if value is not None: log.warning("LogDenied '%s' is invalid, using default value '%s'", value, config.FALLBACK_LOG_DENIED) self.set("LogDenied", str(config.FALLBACK_LOG_DENIED)) # check automatic helpers value = self.get("AutomaticHelpers") if not value or value.lower() not in config.AUTOMATIC_HELPERS_VALUES: if value is not None: log.warning("AutomaticHelpers '%s' is not valid, using default " "value %s", value if value else '', config.FALLBACK_AUTOMATIC_HELPERS) self.set("AutomaticHelpers", str(config.FALLBACK_AUTOMATIC_HELPERS)) value = self.get("FirewallBackend") if not value or value.lower() not in config.FIREWALL_BACKEND_VALUES: if value is not None: log.warning("FirewallBackend '%s' is not valid, using default " "value %s", value if value else '', config.FALLBACK_FIREWALL_BACKEND) self.set("FirewallBackend", str(config.FALLBACK_FIREWALL_BACKEND)) value = self.get("FlushAllOnReload") if not value or value.lower() not in [ "yes", "true", "no", "false" ]: if value is not None: log.warning("FlushAllOnReload '%s' is not valid, using default " "value %s", value if value else '', config.FALLBACK_FLUSH_ALL_ON_RELOAD) self.set("FlushAllOnReload", str(config.FALLBACK_FLUSH_ALL_ON_RELOAD)) value = self.get("ReloadPolicy") try: value = self._parse_reload_policy(value) except ValueError: log.warning( "ReloadPolicy '%s' is not valid, using default value '%s'", value, config.FALLBACK_RELOAD_POLICY, ) self.set("ReloadPolicy", config.FALLBACK_RELOAD_POLICY) value = self.get("RFC3964_IPv4") if not value or value.lower() not in [ "yes", "true", "no", "false" ]: if value is not None: log.warning("RFC3964_IPv4 '%s' is not valid, using default " "value %s", value if value else '', config.FALLBACK_RFC3964_IPV4) self.set("RFC3964_IPv4", str(config.FALLBACK_RFC3964_IPV4)) value = self.get("AllowZoneDrifting") if not value or value.lower() not in [ "yes", "true", "no", "false" ]: if value is not None: log.warning("AllowZoneDrifting '%s' is not valid, using default " "value %s", value if value else '', config.FALLBACK_ALLOW_ZONE_DRIFTING) self.set("AllowZoneDrifting", "yes" if config.FALLBACK_ALLOW_ZONE_DRIFTING else "no") self.sanity_check() # save to self.filename if there are key/value changes def write(self): if len(self._config) < 1: # no changes: nothing to do return # handled keys done = [ ] if not os.path.exists(config.ETC_FIREWALLD): os.mkdir(config.ETC_FIREWALLD, 0o750) try: temp_file = tempfile.NamedTemporaryFile(mode='wt', prefix="%s." % os.path.basename(self.filename), dir=os.path.dirname(self.filename), delete=False) except Exception as msg: log.error("Failed to open temporary file: %s" % msg) raise modified = False empty = False try: f= io.open(self.filename, mode='rt', encoding='UTF-8') except Exception as msg: if os.path.exists(self.filename): log.error("Failed to open '%s': %s" % (self.filename, msg)) raise else: f = None else: for line in f: if not line: break # remove newline line = line.strip("\n") if len(line) < 1: if not empty: temp_file.write(u"\n") empty = True elif line[0] == '#': empty = False temp_file.write(line) temp_file.write(u"\n") else: p = line.split("=") if len(p) != 2: empty = False temp_file.write(line+u"\n") continue key = p[0].strip() value = p[1].strip() # check for modified key/value pairs if key not in done: if (key in self._config and \ self._config[key] != value): empty = False temp_file.write(u'%s=%s\n' % (key, self._config[key])) modified = True elif key in self._deleted: modified = True else: empty = False temp_file.write(line+u"\n") done.append(key) else: modified = True # write remaining key/value pairs if len(self._config) > 0: for (key,value) in self._config.items(): if key in done: continue if key in ["MinimalMark", "AutomaticHelpers", "AllowZoneDrifting"]: # omit deprecated from new config continue if not empty: temp_file.write(u"\n") empty = True temp_file.write(u'%s=%s\n' % (key, value)) modified = True if f: f.close() temp_file.close() if not modified: # not modified: remove tempfile os.remove(temp_file.name) return # make backup if os.path.exists(self.filename): try: shutil.copy2(self.filename, "%s.old" % self.filename) except Exception as msg: os.remove(temp_file.name) raise IOError("Backup of '%s' failed: %s" % (self.filename, msg)) # copy tempfile try: shutil.move(temp_file.name, self.filename) except Exception as msg: os.remove(temp_file.name) raise IOError("Failed to create '%s': %s" % (self.filename, msg)) else: os.chmod(self.filename, 0o600) @staticmethod def _parse_reload_policy(value): valid = True result = { "INPUT": "DROP", "FORWARD": "DROP", "OUTPUT": "DROP", } if value: value = value.strip() v = value.upper() if v in ("ACCEPT", "REJECT", "DROP"): for k in result: result[k] = v else: for a in value.replace(";", ",").split(","): a = a.strip() if not a: continue a2 = a.replace("=", ":").split(":", 2) if len(a2) != 2: valid = False continue k = a2[0].strip().upper() if k not in result: valid = False continue v = a2[1].strip().upper() if v not in ("ACCEPT", "REJECT", "DROP"): valid = False continue result[k] = v if not valid: raise ValueError("Invalid ReloadPolicy") return result @staticmethod def _unparse_reload_policy(value): return ",".join(f"{k}:{v}" for k, v in value.items())	15.87 KB	2026-03-10 23:36
📄 functions.py # -- coding: utf-8 -- # # Copyright (C) 2018 Red Hat, Inc. # # Authors: # Eric Garver <egarver@redhat.com> # # This program is free software; you can redistribute it and/or modify # it under the terms of the GNU General Public License as published by # the Free Software Foundation; either version 2 of the License, or # (at your option) any later version. # # This program is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU General Public License for more details. # # You should have received a copy of the GNU General Public License # along with this program. If not, see <http://www.gnu.org/licenses/>. # import os from firewall import config from firewall.errors import FirewallError from firewall.core.fw_config import FirewallConfig from firewall.core.io.zone import zone_reader from firewall.core.io.service import service_reader from firewall.core.io.ipset import ipset_reader from firewall.core.io.icmptype import icmptype_reader from firewall.core.io.helper import helper_reader from firewall.core.io.policy import policy_reader from firewall.core.io.direct import Direct from firewall.core.io.lockdown_whitelist import LockdownWhitelist from firewall.core.io.firewalld_conf import firewalld_conf def check_on_disk_config(fw): fw_config = FirewallConfig(fw) try: _firewalld_conf = firewalld_conf(config.FIREWALLD_CONF) _firewalld_conf.read() except FirewallError as error: raise FirewallError(error.code, "'%s': %s" % (config.FIREWALLD_CONF, error.msg)) except IOError: # defaults will be filled pass except Exception as msg: raise Exception("'%s': %s" % (config.FIREWALLD_CONF, msg)) fw_config.set_firewalld_conf(_firewalld_conf) readers = { "ipset": { "reader": ipset_reader, "add": fw_config.add_ipset, "dirs": [config.FIREWALLD_IPSETS, config.ETC_FIREWALLD_IPSETS], }, "helper": { "reader": helper_reader, "add": fw_config.add_helper, "dirs": [config.FIREWALLD_HELPERS, config.ETC_FIREWALLD_HELPERS], }, "icmptype": { "reader": icmptype_reader, "add": fw_config.add_icmptype, "dirs": [config.FIREWALLD_ICMPTYPES, config.ETC_FIREWALLD_ICMPTYPES], }, "service": { "reader": service_reader, "add": fw_config.add_service, "dirs": [config.FIREWALLD_SERVICES, config.ETC_FIREWALLD_SERVICES], }, "zone": { "reader": zone_reader, "add": fw_config.add_zone, "dirs": [config.FIREWALLD_ZONES, config.ETC_FIREWALLD_ZONES], }, "policy": { "reader": policy_reader, "add": fw_config.add_policy_object, "dirs": [config.FIREWALLD_POLICIES, config.ETC_FIREWALLD_POLICIES], }, } for reader in readers.keys(): for _dir in readers[reader]["dirs"]: if not os.path.isdir(_dir): continue for file in sorted(os.listdir(_dir)): if file.endswith(".xml"): obj = readers[reader]["reader"](file, _dir) readers[reader]["add"](obj) fw_config.full_check_config() if os.path.isfile(config.FIREWALLD_DIRECT): try: obj = Direct(config.FIREWALLD_DIRECT) obj.read() obj.check_config(obj.export_config()) except FirewallError as error: raise FirewallError(error.code, "'%s': %s" % (config.FIREWALLD_DIRECT, error.msg)) except Exception as msg: raise Exception("'%s': %s" % (config.FIREWALLD_DIRECT, msg)) if os.path.isfile(config.LOCKDOWN_WHITELIST): try: obj = LockdownWhitelist(config.LOCKDOWN_WHITELIST) obj.read() obj.check_config(obj.export_config()) except FirewallError as error: raise FirewallError(error.code, "'%s': %s" % (config.LOCKDOWN_WHITELIST, error.msg)) except Exception as msg: raise Exception("'%s': %s" % (config.LOCKDOWN_WHITELIST, msg))	4.16 KB	2026-03-10 23:36
📄 helper.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__ = [ "Helper", "helper_reader", "helper_writer" ] import xml.sax as sax import os import io import shutil from firewall import config from firewall.core.io.io_object import IO_Object, \ IO_Object_ContentHandler, IO_Object_XMLGenerator, check_port, \ check_tcpudp from firewall.core.logger import log from firewall import errors from firewall.errors import FirewallError class Helper(IO_Object): IMPORT_EXPORT_STRUCTURE = ( ( "version", "" ), # s ( "short", "" ), # s ( "description", "" ), # s ( "family", "", ), # s ( "module", "", ), # s ( "ports", [ ( "", "" ), ], ), # a(ss) ) DBUS_SIGNATURE = '(sssssa(ss))' ADDITIONAL_ALNUM_CHARS = [ "-", "." ] PARSER_REQUIRED_ELEMENT_ATTRS = { "short": None, "description": None, "helper": [ "module" ], } PARSER_OPTIONAL_ELEMENT_ATTRS = { "helper": [ "name", "version", "family" ], "port": [ "port", "protocol" ], } def __init__(self): super(Helper, self).__init__() self.version = "" self.short = "" self.description = "" self.module = "" self.family = "" self.ports = [ ] def cleanup(self): self.version = "" self.short = "" self.description = "" self.module = "" self.family = "" del self.ports[:] def check_ipv(self, ipv): ipvs = [ 'ipv4', 'ipv6' ] if ipv not in ipvs: raise FirewallError(errors.INVALID_IPV, "'%s' not in '%s'" % (ipv, ipvs)) def _check_config(self, config, item, all_config, all_io_objects): if item == "ports": for port in config: check_port(port[0]) check_tcpudp(port[1]) elif item == "module": if not config.startswith("nf_conntrack_"): raise FirewallError( errors.INVALID_MODULE, "'%s' does not start with 'nf_conntrack_'" % config) if len(config.replace("nf_conntrack_", "")) < 1: raise FirewallError(errors.INVALID_MODULE, "Module name '%s' too short" % config) # PARSER class helper_ContentHandler(IO_Object_ContentHandler): def startElement(self, name, attrs): IO_Object_ContentHandler.startElement(self, name, attrs) self.item.parser_check_element_attrs(name, attrs) if name == "helper": if "version" in attrs: self.item.version = attrs["version"] if "family" in attrs: self.item.check_ipv(attrs["family"]) self.item.family = attrs["family"] if "module" in attrs: if not attrs["module"].startswith("nf_conntrack_"): raise FirewallError( errors.INVALID_MODULE, "'%s' does not start with 'nf_conntrack_'" % \ attrs["module"]) if len(attrs["module"].replace("nf_conntrack_", "")) < 1: raise FirewallError( errors.INVALID_MODULE, "Module name '%s' too short" % attrs["module"]) self.item.module = attrs["module"] elif name == "short": pass elif name == "description": pass elif name == "port": check_port(attrs["port"]) check_tcpudp(attrs["protocol"]) entry = (attrs["port"], attrs["protocol"]) if entry not in self.item.ports: self.item.ports.append(entry) else: log.warning("Port '%s/%s' already set, ignoring.", attrs["port"], attrs["protocol"]) def helper_reader(filename, path): helper = Helper() if not filename.endswith(".xml"): raise FirewallError(errors.INVALID_NAME, "'%s' is missing .xml suffix" % filename) helper.name = filename[:-4] helper.check_name(helper.name) helper.filename = filename helper.path = path helper.builtin = False if path.startswith(config.ETC_FIREWALLD) else True helper.default = helper.builtin handler = helper_ContentHandler(helper) parser = sax.make_parser() parser.setContentHandler(handler) name = "%s/%s" % (path, filename) with open(name, "rb") as f: source = sax.InputSource(None) source.setByteStream(f) try: parser.parse(source) except sax.SAXParseException as msg: raise FirewallError(errors.INVALID_HELPER, "not a valid helper file: %s" % \ msg.getException()) del handler del parser return helper def helper_writer(helper, path=None): _path = path if path else helper.path if helper.filename: name = "%s/%s" % (_path, helper.filename) else: name = "%s/%s.xml" % (_path, helper.name) if os.path.exists(name): try: shutil.copy2(name, "%s.old" % name) except Exception as msg: log.error("Backup of file '%s' failed: %s", name, msg) dirpath = os.path.dirname(name) if dirpath.startswith(config.ETC_FIREWALLD) and not os.path.exists(dirpath): if not os.path.exists(config.ETC_FIREWALLD): os.mkdir(config.ETC_FIREWALLD, 0o750) os.mkdir(dirpath, 0o750) f = io.open(name, mode='wt', encoding='UTF-8') handler = IO_Object_XMLGenerator(f) handler.startDocument() # start helper element attrs = {} attrs["module"] = helper.module if helper.version and helper.version != "": attrs["version"] = helper.version if helper.family and helper.family != "": attrs["family"] = helper.family handler.startElement("helper", attrs) handler.ignorableWhitespace("\n") # short if helper.short and helper.short != "": handler.ignorableWhitespace(" ") handler.startElement("short", { }) handler.characters(helper.short) handler.endElement("short") handler.ignorableWhitespace("\n") # description if helper.description and helper.description != "": handler.ignorableWhitespace(" ") handler.startElement("description", { }) handler.characters(helper.description) handler.endElement("description") handler.ignorableWhitespace("\n") # ports for port in helper.ports: handler.ignorableWhitespace(" ") handler.simpleElement("port", { "port": port[0], "protocol": port[1] }) handler.ignorableWhitespace("\n") # end helper element handler.endElement('helper') handler.ignorableWhitespace("\n") handler.endDocument() f.close() del handler	7.55 KB	2026-03-10 23:36
📄 icmptype.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__ = [ "IcmpType", "icmptype_reader", "icmptype_writer" ] import xml.sax as sax import os import io import shutil from firewall import config from firewall.core.io.io_object import IO_Object, \ IO_Object_ContentHandler, IO_Object_XMLGenerator from firewall.core.logger import log from firewall import errors from firewall.errors import FirewallError class IcmpType(IO_Object): IMPORT_EXPORT_STRUCTURE = ( ( "version", "" ), # s ( "short", "" ), # s ( "description", "" ), # s ( "destination", [ "", ], ), # as ) DBUS_SIGNATURE = '(sssas)' ADDITIONAL_ALNUM_CHARS = [ "_", "-" ] PARSER_REQUIRED_ELEMENT_ATTRS = { "short": None, "description": None, "icmptype": None, } PARSER_OPTIONAL_ELEMENT_ATTRS = { "icmptype": [ "name", "version" ], "destination": [ "ipv4", "ipv6" ], } def __init__(self): super(IcmpType, self).__init__() self.version = "" self.short = "" self.description = "" self.destination = [ ] def cleanup(self): self.version = "" self.short = "" self.description = "" del self.destination[:] def _check_config(self, config, item, all_config, all_io_objects): if item == "destination": for destination in config: if destination not in [ "ipv4", "ipv6" ]: raise FirewallError(errors.INVALID_DESTINATION, "'%s' not from {'ipv4'\|'ipv6'}" % \ destination) # PARSER class icmptype_ContentHandler(IO_Object_ContentHandler): def startElement(self, name, attrs): IO_Object_ContentHandler.startElement(self, name, attrs) self.item.parser_check_element_attrs(name, attrs) if name == "icmptype": if "name" in attrs: log.warning("Ignoring deprecated attribute name='%s'" % attrs["name"]) if "version" in attrs: self.item.version = attrs["version"] elif name == "short": pass elif name == "description": pass elif name == "destination": for x in [ "ipv4", "ipv6" ]: if x in attrs and \ attrs[x].lower() in [ "yes", "true" ]: self.item.destination.append(str(x)) def icmptype_reader(filename, path): icmptype = IcmpType() if not filename.endswith(".xml"): raise FirewallError(errors.INVALID_NAME, "%s is missing .xml suffix" % filename) icmptype.name = filename[:-4] icmptype.check_name(icmptype.name) icmptype.filename = filename icmptype.path = path icmptype.builtin = False if path.startswith(config.ETC_FIREWALLD) else True icmptype.default = icmptype.builtin handler = icmptype_ContentHandler(icmptype) parser = sax.make_parser() parser.setContentHandler(handler) name = "%s/%s" % (path, filename) with open(name, "rb") as f: source = sax.InputSource(None) source.setByteStream(f) try: parser.parse(source) except sax.SAXParseException as msg: raise FirewallError(errors.INVALID_ICMPTYPE, "not a valid icmptype file: %s" % \ msg.getException()) del handler del parser return icmptype def icmptype_writer(icmptype, path=None): _path = path if path else icmptype.path if icmptype.filename: name = "%s/%s" % (_path, icmptype.filename) else: name = "%s/%s.xml" % (_path, icmptype.name) if os.path.exists(name): try: shutil.copy2(name, "%s.old" % name) except Exception as msg: log.error("Backup of file '%s' failed: %s", name, msg) dirpath = os.path.dirname(name) if dirpath.startswith(config.ETC_FIREWALLD) and not os.path.exists(dirpath): if not os.path.exists(config.ETC_FIREWALLD): os.mkdir(config.ETC_FIREWALLD, 0o750) os.mkdir(dirpath, 0o750) f = io.open(name, mode='wt', encoding='UTF-8') handler = IO_Object_XMLGenerator(f) handler.startDocument() # start icmptype element attrs = {} if icmptype.version and icmptype.version != "": attrs["version"] = icmptype.version handler.startElement("icmptype", attrs) handler.ignorableWhitespace("\n") # short if icmptype.short and icmptype.short != "": handler.ignorableWhitespace(" ") handler.startElement("short", { }) handler.characters(icmptype.short) handler.endElement("short") handler.ignorableWhitespace("\n") # description if icmptype.description and icmptype.description != "": handler.ignorableWhitespace(" ") handler.startElement("description", { }) handler.characters(icmptype.description) handler.endElement("description") handler.ignorableWhitespace("\n") # destination if icmptype.destination: handler.ignorableWhitespace(" ") attrs = { } for x in icmptype.destination: attrs[x] = "yes" handler.simpleElement("destination", attrs) handler.ignorableWhitespace("\n") # end icmptype element handler.endElement('icmptype') handler.ignorableWhitespace("\n") handler.endDocument() f.close() del handler	6.13 KB	2026-03-10 23:36
📄 ifcfg.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/>. # """ifcfg file parser""" __all__ = [ "ifcfg" ] import os.path import io import tempfile import shutil from firewall.core.logger import log class ifcfg(object): def __init__(self, filename): self._config = { } self._deleted = [ ] self.filename = filename self.clear() def clear(self): self._config = { } self._deleted = [ ] def cleanup(self): self._config.clear() def get(self, key): return self._config.get(key.strip()) def set(self, key, value): _key = key.strip() self._config[_key] = value.strip() if _key in self._deleted: self._deleted.remove(_key) def __str__(self): s = "" for (key, value) in self._config.items(): if s: s += '\n' s += '%s=%s' % (key, value) return s # load self.filename def read(self): self.clear() try: f = open(self.filename, "r") except Exception as msg: log.error("Failed to load '%s': %s", self.filename, msg) raise for line in f: if not line: break line = line.strip() if len(line) < 1 or line[0] in ['#', ';']: continue # get key/value pair pair = [ x.strip() for x in line.split("=", 1) ] if len(pair) != 2: continue if len(pair[1]) >= 2 and \ pair[1].startswith('"') and pair[1].endswith('"'): pair[1] = pair[1][1:-1] if pair[1] == '': continue elif self._config.get(pair[0]) is not None: log.warning("%s: Duplicate option definition: '%s'", self.filename, line.strip()) continue self._config[pair[0]] = pair[1] f.close() def write(self): if len(self._config) < 1: # no changes: nothing to do return # handled keys done = [ ] try: temp_file = tempfile.NamedTemporaryFile( mode='wt', prefix="%s." % os.path.basename(self.filename), dir=os.path.dirname(self.filename), delete=False) except Exception as msg: log.error("Failed to open temporary file: %s" % msg) raise modified = False empty = False try: f = io.open(self.filename, mode='rt', encoding='UTF-8') except Exception as msg: if os.path.exists(self.filename): log.error("Failed to open '%s': %s" % (self.filename, msg)) raise else: f = None else: for line in f: if not line: break # remove newline line = line.strip("\n") if len(line) < 1: if not empty: temp_file.write(u"\n") empty = True elif line[0] == '#': empty = False temp_file.write(line) temp_file.write(u"\n") else: p = line.split("=", 1) if len(p) != 2: empty = False temp_file.write(line+u"\n") continue key = p[0].strip() value = p[1].strip() if len(value) >= 2 and \ value.startswith('"') and value.endswith('"'): value = value[1:-1] # check for modified key/value pairs if key not in done: if key in self._config and self._config[key] != value: empty = False temp_file.write(u'%s=%s\n' % (key, self._config[key])) modified = True elif key in self._deleted: modified = True else: empty = False temp_file.write(line+u"\n") done.append(key) else: modified = True # write remaining key/value pairs if len(self._config) > 0: for (key, value) in self._config.items(): if key in done: continue if not empty: empty = True temp_file.write(u'%s=%s\n' % (key, value)) modified = True if f: f.close() temp_file.close() if not modified: # not modified: remove tempfile os.remove(temp_file.name) return # make backup if os.path.exists(self.filename): try: shutil.copy2(self.filename, "%s.bak" % self.filename) except Exception as msg: os.remove(temp_file.name) raise IOError("Backup of '%s' failed: %s" % (self.filename, msg)) # copy tempfile try: shutil.move(temp_file.name, self.filename) except Exception as msg: os.remove(temp_file.name) raise IOError("Failed to create '%s': %s" % (self.filename, msg)) else: os.chmod(self.filename, 0o600)	6.15 KB	2026-03-10 23:36
📄 io_object.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/>. # """Generic io_object handler, io specific check methods.""" __all__ = [ "IO_Object", "IO_Object_ContentHandler", "IO_Object_XMLGenerator", "check_port", "check_tcpudp", "check_protocol", "check_address" ] import xml.sax as sax import xml.sax.saxutils as saxutils import copy from collections import OrderedDict from firewall import functions from firewall import errors from firewall.errors import FirewallError class IO_Object(object): """ Abstract IO_Object as base for icmptype, service and zone """ IMPORT_EXPORT_STRUCTURE = ( ) DBUS_SIGNATURE = '()' ADDITIONAL_ALNUM_CHARS = [ ] # additional to alnum PARSER_REQUIRED_ELEMENT_ATTRS = { } PARSER_OPTIONAL_ELEMENT_ATTRS = { } def __init__(self): self.filename = "" self.path = "" self.name = "" self.default = False self.builtin = False def export_config(self): ret = [ ] for x in self.IMPORT_EXPORT_STRUCTURE: ret.append(copy.deepcopy(getattr(self, x[0]))) return tuple(ret) def export_config_dict(self): conf = {} type_formats = dict([(x[0], x[1]) for x in self.IMPORT_EXPORT_STRUCTURE]) for key in type_formats: if getattr(self, key) or isinstance(getattr(self, key), bool): conf[key] = copy.deepcopy(getattr(self, key)) return conf def import_config(self, conf, all_io_objects): self.check_config(conf, all_io_objects) for i,(element,dummy) in enumerate(self.IMPORT_EXPORT_STRUCTURE): if isinstance(conf[i], list): # remove duplicates without changing the order _conf = [ ] _set = set() for x in conf[i]: if x not in _set: _conf.append(x) _set.add(x) del _set setattr(self, element, copy.deepcopy(_conf)) else: setattr(self, element, copy.deepcopy(conf[i])) def import_config_dict(self, conf, all_io_objects): self.check_config_dict(conf, all_io_objects) for key in conf: if not hasattr(self, key): raise FirewallError(errors.UNKNOWN_ERROR, "Internal error. '{}' is not a valid attribute".format(key)) if isinstance(conf[key], list): # maintain list order while removing duplicates setattr(self, key, list(OrderedDict.fromkeys(copy.deepcopy(conf[key])))) else: setattr(self, key, copy.deepcopy(conf[key])) def check_name(self, name): if not isinstance(name, str): raise FirewallError(errors.INVALID_TYPE, "'%s' not of type %s, but %s" % (name, type(""), type(name))) if len(name) < 1: raise FirewallError(errors.INVALID_NAME, "name can't be empty") for char in name: if not char.isalnum() and char not in self.ADDITIONAL_ALNUM_CHARS: raise FirewallError( errors.INVALID_NAME, "'%s' is not allowed in '%s'" % ((char, name))) def check_config(self, conf, all_io_objects={}): if len(conf) != len(self.IMPORT_EXPORT_STRUCTURE): raise FirewallError( errors.INVALID_TYPE, "structure size mismatch %d != %d" % \ (len(conf), len(self.IMPORT_EXPORT_STRUCTURE))) conf_dict = {} for i,(x,y) in enumerate(self.IMPORT_EXPORT_STRUCTURE): conf_dict[x] = conf[i] self.check_config_dict(conf_dict, all_io_objects) def check_config_dict(self, conf, all_io_objects): type_formats = dict([(x[0], x[1]) for x in self.IMPORT_EXPORT_STRUCTURE]) for key in conf: if key not in [x for (x,y) in self.IMPORT_EXPORT_STRUCTURE]: raise FirewallError(errors.INVALID_OPTION, "option '{}' is not valid".format(key)) self._check_config_structure(conf[key], type_formats[key]) self._check_config(conf[key], key, conf, all_io_objects) def _check_config(self, dummy1, dummy2, dummy3, dummy4): # to be overloaded by sub classes return def _check_config_structure(self, conf, structure): if not isinstance(conf, type(structure)): raise FirewallError(errors.INVALID_TYPE, "'%s' not of type %s, but %s" % \ (conf, type(structure), type(conf))) if isinstance(structure, list): # same type elements, else struct if len(structure) != 1: raise FirewallError(errors.INVALID_TYPE, "len('%s') != 1" % structure) for x in conf: self._check_config_structure(x, structure[0]) elif isinstance(structure, tuple): if len(structure) != len(conf): raise FirewallError(errors.INVALID_TYPE, "len('%s') != %d" % (conf, len(structure))) for i,value in enumerate(structure): self._check_config_structure(conf[i], value) elif isinstance(structure, dict): # only one key value pair in structure (skey, svalue) = list(structure.items())[0] for (key, value) in conf.items(): if not isinstance(key, type(skey)): raise FirewallError(errors.INVALID_TYPE, "'%s' not of type %s, but %s" % (\ key, type(skey), type(key))) if not isinstance(value, type(svalue)): raise FirewallError(errors.INVALID_TYPE, "'%s' not of type %s, but %s" % (\ value, type(svalue), type(value))) # check required elements and attributes and also optional attributes def parser_check_element_attrs(self, name, attrs): _attrs = attrs.getNames() found = False if name in self.PARSER_REQUIRED_ELEMENT_ATTRS: found = True if self.PARSER_REQUIRED_ELEMENT_ATTRS[name] is not None: for x in self.PARSER_REQUIRED_ELEMENT_ATTRS[name]: if x in _attrs: _attrs.remove(x) else: raise FirewallError( errors.PARSE_ERROR, "Missing attribute %s for %s" % (x, name)) if name in self.PARSER_OPTIONAL_ELEMENT_ATTRS: found = True for x in self.PARSER_OPTIONAL_ELEMENT_ATTRS[name]: if x in _attrs: _attrs.remove(x) if not found: raise FirewallError(errors.PARSE_ERROR, "Unexpected element %s" % name) # raise attributes[0] for x in _attrs: raise FirewallError(errors.PARSE_ERROR, "%s: Unexpected attribute %s" % (name, x)) # PARSER class UnexpectedElementError(Exception): def __init__(self, name): super(UnexpectedElementError, self).__init__() self.name = name def __str__(self): return "Unexpected element '%s'" % (self.name) class MissingAttributeError(Exception): def __init__(self, name, attribute): super(MissingAttributeError, self).__init__() self.name = name self.attribute = attribute def __str__(self): return "Element '%s': missing '%s' attribute" % \ (self.name, self.attribute) class UnexpectedAttributeError(Exception): def __init__(self, name, attribute): super(UnexpectedAttributeError, self).__init__() self.name = name self.attribute = attribute def __str__(self): return "Element '%s': unexpected attribute '%s'" % \ (self.name, self.attribute) class IO_Object_ContentHandler(sax.handler.ContentHandler): def __init__(self, item): self.item = item self._element = "" def startDocument(self): self._element = "" def startElement(self, name, attrs): self._element = "" def endElement(self, name): if name == "short": self.item.short = self._element elif name == "description": self.item.description = self._element def characters(self, content): self._element += content.replace('\n', ' ') class IO_Object_XMLGenerator(saxutils.XMLGenerator): def __init__(self, out): # fix memory leak in saxutils.XMLGenerator.__init__: # out = _gettextwriter(out, encoding) # creates unbound object results in garbage in gc # # saxutils.XMLGenerator.__init__(self, out, "utf-8") # replaced by modified saxutils.XMLGenerator.__init__ code: sax.handler.ContentHandler.__init__(self) self._write = out.write self._flush = out.flush self._ns_contexts = [{}] # contains uri -> prefix dicts self._current_context = self._ns_contexts[-1] self._undeclared_ns_maps = [] self._encoding = "utf-8" self._pending_start_element = False self._short_empty_elements = False def simpleElement(self, name, attrs): """ slightly modified startElement() """ self._write('<' + name) for (name, value) in attrs.items(): self._write(' %s=%s' % (name, saxutils.quoteattr(value))) self._write('/>') def check_port(port): port_range = functions.getPortRange(port) if port_range == -2: raise FirewallError(errors.INVALID_PORT, "port number in '%s' is too big" % port) elif port_range == -1: raise FirewallError(errors.INVALID_PORT, "'%s' is invalid port range" % port) elif port_range is None: raise FirewallError(errors.INVALID_PORT, "port range '%s' is ambiguous" % port) elif len(port_range) == 2 and port_range[0] >= port_range[1]: raise FirewallError(errors.INVALID_PORT, "'%s' is invalid port range" % port) def check_tcpudp(protocol): if protocol not in [ "tcp", "udp", "sctp", "dccp" ]: raise FirewallError(errors.INVALID_PROTOCOL, "'%s' not from {'tcp'\|'udp'\|'sctp'\|'dccp'}" % \ protocol) def check_protocol(protocol): if not functions.checkProtocol(protocol): raise FirewallError(errors.INVALID_PROTOCOL, protocol) def check_address(ipv, addr): if not functions.check_address(ipv, addr): raise FirewallError(errors.INVALID_ADDR, "'%s' is not valid %s address" % (addr, ipv))	11.5 KB	2026-03-10 23:36
📄 ipset.py # -- coding: utf-8 -- # # Copyright (C) 2015-2016 Red Hat, Inc. # # Authors: # Thomas Woerner <twoerner@redhat.com> # # This program is free software; you can redistribute it and/or modify # it under the terms of the GNU General Public License as published by # the Free Software Foundation; either version 2 of the License, or # (at your option) any later version. # # This program is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU General Public License for more details. # # You should have received a copy of the GNU General Public License # along with this program. If not, see <http://www.gnu.org/licenses/>. # """ipset io XML handler, reader, writer""" __all__ = [ "IPSet", "ipset_reader", "ipset_writer" ] import xml.sax as sax import os import io import shutil from firewall import config from firewall.functions import checkIP, checkIP6, checkIPnMask, \ checkIP6nMask, check_mac, check_port, checkInterface, \ checkProtocol from firewall.core.io.io_object import IO_Object, \ IO_Object_ContentHandler, IO_Object_XMLGenerator from firewall.core.ipset import IPSET_TYPES, IPSET_CREATE_OPTIONS from firewall.core.icmp import check_icmp_name, check_icmp_type_code, \ check_icmpv6_name, check_icmpv6_type_code from firewall.core.logger import log from firewall import errors from firewall.errors import FirewallError class IPSet(IO_Object): IMPORT_EXPORT_STRUCTURE = ( ( "version", "" ), # s ( "short", "" ), # s ( "description", "" ), # s ( "type", "" ), # s ( "options", { "": "", }, ), # a{ss} ( "entries", [ "" ], ), # as ) DBUS_SIGNATURE = '(ssssa{ss}as)' ADDITIONAL_ALNUM_CHARS = [ "_", "-", ":", "." ] PARSER_REQUIRED_ELEMENT_ATTRS = { "short": None, "description": None, "ipset": [ "type" ], "option": [ "name" ], "entry": None, } PARSER_OPTIONAL_ELEMENT_ATTRS = { "ipset": [ "version" ], "option": [ "value" ], } def __init__(self): super(IPSet, self).__init__() self.version = "" self.short = "" self.description = "" self.type = "" self.entries = [ ] self.options = { } self.applied = False def cleanup(self): self.version = "" self.short = "" self.description = "" self.type = "" del self.entries[:] self.options.clear() self.applied = False @staticmethod def check_entry(entry, options, ipset_type): family = "ipv4" if "family" in options: if options["family"] == "inet6": family = "ipv6" if not ipset_type.startswith("hash:"): raise FirewallError(errors.INVALID_IPSET, "ipset type '%s' not usable" % ipset_type) flags = ipset_type[5:].split(",") items = entry.split(",") if len(flags) != len(items) or len(flags) < 1: raise FirewallError( errors.INVALID_ENTRY, "entry '%s' does not match ipset type '%s'" % \ (entry, ipset_type)) for i, flag in enumerate(flags): item = items[i] if flag == "ip": if "-" in item and family == "ipv4": # IP ranges only with plain IPs, no masks if i > 1: raise FirewallError( errors.INVALID_ENTRY, "invalid address '%s' in '%s'[%d]" % \ (item, entry, i)) splits = item.split("-") if len(splits) != 2: raise FirewallError( errors.INVALID_ENTRY, "invalid address range '%s' in '%s' for %s (%s)" % \ (item, entry, ipset_type, family)) for _split in splits: if (family == "ipv4" and not checkIP(_split)) or \ (family == "ipv6" and not checkIP6(_split)): raise FirewallError( errors.INVALID_ENTRY, "invalid address '%s' in '%s' for %s (%s)" % \ (_split, entry, ipset_type, family)) else: # IPs with mask only allowed in the first # position of the type if family == "ipv4": if item == "0.0.0.0": raise FirewallError( errors.INVALID_ENTRY, "invalid address '%s' in '%s' for %s (%s)" % \ (item, entry, ipset_type, family)) if i == 0: ip_check = checkIPnMask else: ip_check = checkIP else: ip_check = checkIP6 if not ip_check(item): raise FirewallError( errors.INVALID_ENTRY, "invalid address '%s' in '%s' for %s (%s)" % \ (item, entry, ipset_type, family)) elif flag == "net": if "-" in item: # IP ranges only with plain IPs, no masks splits = item.split("-") if len(splits) != 2: raise FirewallError( errors.INVALID_ENTRY, "invalid address range '%s' in '%s' for %s (%s)" % \ (item, entry, ipset_type, family)) # First part can only be a plain IP if (family == "ipv4" and not checkIP(splits[0])) or \ (family == "ipv6" and not checkIP6(splits[0])): raise FirewallError( errors.INVALID_ENTRY, "invalid address '%s' in '%s' for %s (%s)" % \ (splits[0], entry, ipset_type, family)) # Second part can also have a mask if (family == "ipv4" and not checkIPnMask(splits[1])) or \ (family == "ipv6" and not checkIP6nMask(splits[1])): raise FirewallError( errors.INVALID_ENTRY, "invalid address '%s' in '%s' for %s (%s)" % \ (splits[1], entry, ipset_type, family)) else: # IPs with mask allowed in all positions, but no /0 if item.endswith("/0"): if not (family == "ipv6" and i == 0 and ipset_type == "hash:net,iface"): raise FirewallError( errors.INVALID_ENTRY, "invalid address '%s' in '%s' for %s (%s)" % \ (item, entry, ipset_type, family)) if (family == "ipv4" and not checkIPnMask(item)) or \ (family == "ipv6" and not checkIP6nMask(item)): raise FirewallError( errors.INVALID_ENTRY, "invalid address '%s' in '%s' for %s (%s)" % \ (item, entry, ipset_type, family)) elif flag == "mac": # ipset does not allow to add 00:00:00:00:00:00 if not check_mac(item) or item == "00:00:00:00:00:00": raise FirewallError( errors.INVALID_ENTRY, "invalid mac address '%s' in '%s'" % (item, entry)) elif flag == "port": if ":" in item: splits = item.split(":") if len(splits) != 2: raise FirewallError( errors.INVALID_ENTRY, "invalid port '%s'" % (item)) if splits[0] == "icmp": if family != "ipv4": raise FirewallError( errors.INVALID_ENTRY, "invalid protocol for family '%s' in '%s'" % \ (family, entry)) if not check_icmp_name(splits[1]) and "/" not in splits[1]: raise FirewallError( errors.INVALID_ENTRY, "invalid icmp type '%s' in '%s'" % \ (splits[1], entry)) else: (_type, _code) = splits[1].split("/") if not check_icmp_type_code(_type, _code): raise FirewallError( errors.INVALID_ENTRY, "invalid icmp type '%s' in '%s'" % (splits[1], entry), ) elif splits[0] in [ "icmpv6", "ipv6-icmp" ]: if family != "ipv6": raise FirewallError( errors.INVALID_ENTRY, "invalid protocol for family '%s' in '%s'" % \ (family, entry)) if not check_icmpv6_name(splits[1]) and "/" not in splits[1]: raise FirewallError( errors.INVALID_ENTRY, "invalid icmpv6 type '%s' in '%s'" % \ (splits[1], entry)) else: (_type, _code) = splits[1].split("/") if not check_icmpv6_type_code(_type, _code): raise FirewallError( errors.INVALID_ENTRY, "invalid icmpv6 type '%s' in '%s'" % (splits[1], entry), ) elif splits[0] not in [ "tcp", "sctp", "udp", "udplite" ] \ and not checkProtocol(splits[0]): raise FirewallError( errors.INVALID_ENTRY, "invalid protocol '%s' in '%s'" % (splits[0], entry)) elif not check_port(splits[1]): raise FirewallError( errors.INVALID_ENTRY, "invalid port '%s'in '%s'" % (splits[1], entry)) else: if not check_port(item): raise FirewallError( errors.INVALID_ENTRY, "invalid port '%s' in '%s'" % (item, entry)) elif flag == "mark": if item.startswith("0x"): try: int_val = int(item, 16) except ValueError: raise FirewallError( errors.INVALID_ENTRY, "invalid mark '%s' in '%s'" % (item, entry)) else: try: int_val = int(item) except ValueError: raise FirewallError( errors.INVALID_ENTRY, "invalid mark '%s' in '%s'" % (item, entry)) if int_val < 0 or int_val > 4294967295: raise FirewallError( errors.INVALID_ENTRY, "invalid mark '%s' in '%s'" % (item, entry)) elif flag == "iface": if not checkInterface(item) or len(item) > 15: raise FirewallError( errors.INVALID_ENTRY, "invalid interface '%s' in '%s'" % (item, entry)) else: raise FirewallError(errors.INVALID_IPSET, "ipset type '%s' not usable" % ipset_type) def _check_config(self, config, item, all_config, all_io_objects): if item == "type": if config not in IPSET_TYPES: raise FirewallError(errors.INVALID_TYPE, "'%s' is not valid ipset type" % config) if item == "options": for key in config.keys(): if key not in IPSET_CREATE_OPTIONS: raise FirewallError(errors.INVALID_IPSET, "ipset invalid option '%s'" % key) if key in [ "timeout", "hashsize", "maxelem" ]: try: int_value = int(config[key]) except ValueError: raise FirewallError( errors.INVALID_VALUE, "Option '%s': Value '%s' is not an integer" % \ (key, config[key])) if int_value < 0: raise FirewallError( errors.INVALID_VALUE, "Option '%s': Value '%s' is negative" % \ (key, config[key])) elif key == "family" and \ config[key] not in [ "inet", "inet6" ]: raise FirewallError(errors.INVALID_FAMILY, config[key]) def import_config(self, config, all_io_objects): if "timeout" in config[4] and config[4]["timeout"] != "0": if len(config[5]) != 0: raise FirewallError(errors.IPSET_WITH_TIMEOUT) for entry in config[5]: IPSet.check_entry(entry, config[4], config[3]) super(IPSet, self).import_config(config, all_io_objects) # PARSER class ipset_ContentHandler(IO_Object_ContentHandler): def startElement(self, name, attrs): IO_Object_ContentHandler.startElement(self, name, attrs) self.item.parser_check_element_attrs(name, attrs) if name == "ipset": if "type" in attrs: if attrs["type"] not in IPSET_TYPES: raise FirewallError(errors.INVALID_TYPE, "%s" % attrs["type"]) self.item.type = attrs["type"] if "version" in attrs: self.item.version = attrs["version"] elif name == "short": pass elif name == "description": pass elif name == "option": value = "" if "value" in attrs: value = attrs["value"] if attrs["name"] not in \ [ "family", "timeout", "hashsize", "maxelem" ]: raise FirewallError( errors.INVALID_OPTION, "Unknown option '%s'" % attrs["name"]) if self.item.type == "hash:mac" and attrs["name"] in [ "family" ]: raise FirewallError( errors.INVALID_OPTION, "Unsupported option '%s' for type '%s'" % \ (attrs["name"], self.item.type)) if attrs["name"] in [ "family", "timeout", "hashsize", "maxelem" ] \ and not value: raise FirewallError( errors.INVALID_OPTION, "Missing mandatory value of option '%s'" % attrs["name"]) if attrs["name"] in [ "timeout", "hashsize", "maxelem" ]: try: int_value = int(value) except ValueError: raise FirewallError( errors.INVALID_VALUE, "Option '%s': Value '%s' is not an integer" % \ (attrs["name"], value)) if int_value < 0: raise FirewallError( errors.INVALID_VALUE, "Option '%s': Value '%s' is negative" % \ (attrs["name"], value)) if attrs["name"] == "family" and value not in [ "inet", "inet6" ]: raise FirewallError(errors.INVALID_FAMILY, value) if attrs["name"] not in self.item.options: self.item.options[attrs["name"]] = value else: log.warning("Option %s already set, ignoring.", attrs["name"]) # nothing to do for entry and entries here def endElement(self, name): IO_Object_ContentHandler.endElement(self, name) if name == "entry": self.item.entries.append(self._element) def ipset_reader(filename, path): ipset = IPSet() if not filename.endswith(".xml"): raise FirewallError(errors.INVALID_NAME, "'%s' is missing .xml suffix" % filename) ipset.name = filename[:-4] ipset.check_name(ipset.name) ipset.filename = filename ipset.path = path ipset.builtin = False if path.startswith(config.ETC_FIREWALLD) else True ipset.default = ipset.builtin handler = ipset_ContentHandler(ipset) parser = sax.make_parser() parser.setContentHandler(handler) name = "%s/%s" % (path, filename) with open(name, "rb") as f: source = sax.InputSource(None) source.setByteStream(f) try: parser.parse(source) except sax.SAXParseException as msg: raise FirewallError(errors.INVALID_IPSET, "not a valid ipset file: %s" % \ msg.getException()) del handler del parser if "timeout" in ipset.options and ipset.options["timeout"] != "0" and \ len(ipset.entries) > 0: # no entries visible for ipsets with timeout log.warning("ipset '%s': timeout option is set, entries are ignored", ipset.name) del ipset.entries[:] i = 0 entries_set = set() while i < len(ipset.entries): if ipset.entries[i] in entries_set: log.warning("Entry %s already set, ignoring.", ipset.entries[i]) ipset.entries.pop(i) else: try: ipset.check_entry(ipset.entries[i], ipset.options, ipset.type) except FirewallError as e: log.warning("%s, ignoring.", e) ipset.entries.pop(i) else: entries_set.add(ipset.entries[i]) i += 1 del entries_set return ipset def ipset_writer(ipset, path=None): _path = path if path else ipset.path if ipset.filename: name = "%s/%s" % (_path, ipset.filename) else: name = "%s/%s.xml" % (_path, ipset.name) if os.path.exists(name): try: shutil.copy2(name, "%s.old" % name) except Exception as msg: log.error("Backup of file '%s' failed: %s", name, msg) dirpath = os.path.dirname(name) if dirpath.startswith(config.ETC_FIREWALLD) and not os.path.exists(dirpath): if not os.path.exists(config.ETC_FIREWALLD): os.mkdir(config.ETC_FIREWALLD, 0o750) os.mkdir(dirpath, 0o750) f = io.open(name, mode='wt', encoding='UTF-8') handler = IO_Object_XMLGenerator(f) handler.startDocument() # start ipset element attrs = { "type": ipset.type } if ipset.version and ipset.version != "": attrs["version"] = ipset.version handler.startElement("ipset", attrs) handler.ignorableWhitespace("\n") # short if ipset.short and ipset.short != "": handler.ignorableWhitespace(" ") handler.startElement("short", { }) handler.characters(ipset.short) handler.endElement("short") handler.ignorableWhitespace("\n") # description if ipset.description and ipset.description != "": handler.ignorableWhitespace(" ") handler.startElement("description", { }) handler.characters(ipset.description) handler.endElement("description") handler.ignorableWhitespace("\n") # options for key,value in ipset.options.items(): handler.ignorableWhitespace(" ") if value != "": handler.simpleElement("option", { "name": key, "value": value }) else: handler.simpleElement("option", { "name": key }) handler.ignorableWhitespace("\n") # entries for entry in ipset.entries: handler.ignorableWhitespace(" ") handler.startElement("entry", { }) handler.characters(entry) handler.endElement("entry") handler.ignorableWhitespace("\n") # end ipset element handler.endElement('ipset') handler.ignorableWhitespace("\n") handler.endDocument() f.close() del handler	20.8 KB	2026-03-10 23:36
📄 lockdown_whitelist.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/>. # import xml.sax as sax import os import io import shutil from firewall import config from firewall.core.io.io_object import IO_Object, \ IO_Object_ContentHandler, IO_Object_XMLGenerator from firewall.core.logger import log from firewall.functions import uniqify, checkUser, checkUid, checkCommand, \ checkContext from firewall import errors from firewall.errors import FirewallError class lockdown_whitelist_ContentHandler(IO_Object_ContentHandler): def __init__(self, item): IO_Object_ContentHandler.__init__(self, item) self.whitelist = False def startElement(self, name, attrs): IO_Object_ContentHandler.startElement(self, name, attrs) self.item.parser_check_element_attrs(name, attrs) if name == "whitelist": if self.whitelist: raise FirewallError(errors.PARSE_ERROR, "More than one whitelist.") self.whitelist = True elif name == "command": if not self.whitelist: log.error("Parse Error: command outside of whitelist") return command = attrs["name"] self.item.add_command(command) elif name == "user": if not self.whitelist: log.error("Parse Error: user outside of whitelist") return if "id" in attrs: try: uid = int(attrs["id"]) except ValueError: log.error("Parse Error: %s is not a valid uid" % attrs["id"]) return self.item.add_uid(uid) elif "name" in attrs: self.item.add_user(attrs["name"]) elif name == "selinux": if not self.whitelist: log.error("Parse Error: selinux outside of whitelist") return if "context" not in attrs: log.error("Parse Error: no context") return self.item.add_context(attrs["context"]) else: log.error('Unknown XML element %s' % name) return class LockdownWhitelist(IO_Object): """ LockdownWhitelist class """ IMPORT_EXPORT_STRUCTURE = ( ( "commands", [ "" ] ), # as ( "contexts", [ "" ] ), # as ( "users", [ "" ] ), # as ( "uids", [ 0 ] ) # ai ) DBUS_SIGNATURE = '(asasasai)' ADDITIONAL_ALNUM_CHARS = [ "_" ] PARSER_REQUIRED_ELEMENT_ATTRS = { "whitelist": None, "command": [ "name" ], "user": None, # "group": None, "selinux": [ "context" ], } PARSER_OPTIONAL_ELEMENT_ATTRS = { "user": [ "id", "name" ], # "group": [ "id", "name" ], } def __init__(self, filename): super(LockdownWhitelist, self).__init__() self.filename = filename self.parser = None self.commands = [ ] self.contexts = [ ] self.users = [ ] self.uids = [ ] # self.gids = [ ] # self.groups = [ ] def _check_config(self, config, item, all_config, all_io_objects): if item in [ "commands", "contexts", "users", "uids" ]: for x in config: self._check_config(x, item[:-1], all_config, all_io_objects) elif item == "command": if not checkCommand(config): raise FirewallError(errors.INVALID_COMMAND, config) elif item == "context": if not checkContext(config): raise FirewallError(errors.INVALID_CONTEXT, config) elif item == "user": if not checkUser(config): raise FirewallError(errors.INVALID_USER, config) elif item == "uid": if not checkUid(config): raise FirewallError(errors.INVALID_UID, config) def cleanup(self): del self.commands[:] del self.contexts[:] del self.users[:] del self.uids[:] # del self.gids[:] # del self.groups[:] # commands def add_command(self, command): if not checkCommand(command): raise FirewallError(errors.INVALID_COMMAND, command) if command not in self.commands: self.commands.append(command) else: raise FirewallError(errors.ALREADY_ENABLED, 'Command "%s" already in whitelist' % command) def remove_command(self, command): if command in self.commands: self.commands.remove(command) else: raise FirewallError(errors.NOT_ENABLED, 'Command "%s" not in whitelist.' % command) def has_command(self, command): return (command in self.commands) def match_command(self, command): for _command in self.commands: if _command.endswith("*"): if command.startswith(_command[:-1]): return True else: if _command == command: return True return False def get_commands(self): return self.commands # user ids def add_uid(self, uid): if not checkUid(uid): raise FirewallError(errors.INVALID_UID, str(uid)) if uid not in self.uids: self.uids.append(uid) else: raise FirewallError(errors.ALREADY_ENABLED, 'Uid "%s" already in whitelist' % uid) def remove_uid(self, uid): if uid in self.uids: self.uids.remove(uid) else: raise FirewallError(errors.NOT_ENABLED, 'Uid "%s" not in whitelist.' % uid) def has_uid(self, uid): return (uid in self.uids) def match_uid(self, uid): return (uid in self.uids) def get_uids(self): return self.uids # users def add_user(self, user): if not checkUser(user): raise FirewallError(errors.INVALID_USER, user) if user not in self.users: self.users.append(user) else: raise FirewallError(errors.ALREADY_ENABLED, 'User "%s" already in whitelist' % user) def remove_user(self, user): if user in self.users: self.users.remove(user) else: raise FirewallError(errors.NOT_ENABLED, 'User "%s" not in whitelist.' % user) def has_user(self, user): return (user in self.users) def match_user(self, user): return (user in self.users) def get_users(self): return self.users # # group ids # # def add_gid(self, gid): # if gid not in self.gids: # self.gids.append(gid) # # def remove_gid(self, gid): # if gid in self.gids: # self.gids.remove(gid) # else: # raise FirewallError(errors.NOT_ENABLED, # 'Gid "%s" not in whitelist.' % gid) # # def has_gid(self, gid): # return (gid in self.gids) # # def match_gid(self, gid): # return (gid in self.gids) # # def get_gids(self): # return self.gids # # groups # # def add_group(self, group): # if group not in self.groups: # self.groups.append(group) # # def remove_group(self, group): # if group in self.groups: # self.groups.remove(group) # else: # raise FirewallError(errors.NOT_ENABLED, # 'Group "%s" not in whitelist.' % group) # # def has_group(self, group): # return (group in self.groups) # # def match_group(self, group): # return (group in self.groups) # # def get_groups(self): # return self.groups # selinux contexts def add_context(self, context): if not checkContext(context): raise FirewallError(errors.INVALID_CONTEXT, context) if context not in self.contexts: self.contexts.append(context) else: raise FirewallError(errors.ALREADY_ENABLED, 'Context "%s" already in whitelist' % context) def remove_context(self, context): if context in self.contexts: self.contexts.remove(context) else: raise FirewallError(errors.NOT_ENABLED, 'Context "%s" not in whitelist.' % context) def has_context(self, context): return (context in self.contexts) def match_context(self, context): return (context in self.contexts) def get_contexts(self): return self.contexts # read and write def read(self): self.cleanup() if not self.filename.endswith(".xml"): raise FirewallError(errors.INVALID_NAME, "'%s' is missing .xml suffix" % self.filename) handler = lockdown_whitelist_ContentHandler(self) parser = sax.make_parser() parser.setContentHandler(handler) try: parser.parse(self.filename) except sax.SAXParseException as msg: raise FirewallError(errors.INVALID_TYPE, "Not a valid file: %s" % \ msg.getException()) del handler del parser def write(self): if os.path.exists(self.filename): try: shutil.copy2(self.filename, "%s.old" % self.filename) except Exception as msg: raise IOError("Backup of '%s' failed: %s" % (self.filename, msg)) if not os.path.exists(config.ETC_FIREWALLD): os.mkdir(config.ETC_FIREWALLD, 0o750) f = io.open(self.filename, mode='wt', encoding='UTF-8') handler = IO_Object_XMLGenerator(f) handler.startDocument() # start whitelist element handler.startElement("whitelist", { }) handler.ignorableWhitespace("\n") # commands for command in uniqify(self.commands): handler.ignorableWhitespace(" ") handler.simpleElement("command", { "name": command }) handler.ignorableWhitespace("\n") for uid in uniqify(self.uids): handler.ignorableWhitespace(" ") handler.simpleElement("user", { "id": str(uid) }) handler.ignorableWhitespace("\n") for user in uniqify(self.users): handler.ignorableWhitespace(" ") handler.simpleElement("user", { "name": user }) handler.ignorableWhitespace("\n") # for gid in uniqify(self.gids): # handler.ignorableWhitespace(" ") # handler.simpleElement("user", { "id": str(gid) }) # handler.ignorableWhitespace("\n") # for group in uniqify(self.groups): # handler.ignorableWhitespace(" ") # handler.simpleElement("group", { "name": group }) # handler.ignorableWhitespace("\n") for context in uniqify(self.contexts): handler.ignorableWhitespace(" ") handler.simpleElement("selinux", { "context": context }) handler.ignorableWhitespace("\n") # end whitelist element handler.endElement("whitelist") handler.ignorableWhitespace("\n") handler.endDocument() f.close() del handler	11.94 KB	2026-03-10 23:36
📄 policy.py # -- coding: utf-8 -- # # SPDX-License-Identifier: GPL-2.0-or-later __all__ = [ "Policy", "policy_reader", "policy_writer" ] import xml.sax as sax import os import io import shutil from firewall import config from firewall.functions import ( checkIP, checkIP6, checkUINT16, coalescePortRange, max_policy_name_len, portInPortRange, portStr, uniqify, ) from firewall.core.base import DEFAULT_POLICY_TARGET, POLICY_TARGETS, DEFAULT_POLICY_PRIORITY from firewall.core.io.io_object import IO_Object, \ IO_Object_ContentHandler, IO_Object_XMLGenerator, check_port, \ check_tcpudp, check_protocol from firewall.core import rich from firewall.core.logger import log from firewall import errors from firewall.errors import FirewallError def common_startElement(obj, name, attrs): if name == "short": pass elif name == "description": pass elif name == "service": if obj._rule: if obj._rule.element: log.warning("Invalid rule: More than one element in rule '%s', ignoring.", str(obj._rule)) obj._rule_error = True return True obj._rule.element = rich.Rich_Service(attrs["name"]) return True if attrs["name"] not in obj.item.services: obj.item.services.append(attrs["name"]) else: log.warning("Service '%s' already set, ignoring.", attrs["name"]) elif name == "port": if obj._rule: if obj._rule.element: log.warning("Invalid rule: More than one element in rule '%s', ignoring.", str(obj._rule)) obj._rule_error = True return True obj._rule.element = rich.Rich_Port(attrs["port"], attrs["protocol"]) return True check_port(attrs["port"]) check_tcpudp(attrs["protocol"]) # coalesce and warn about overlapping ranges new_port_id = (portStr(attrs["port"], "-"), attrs["protocol"]) existing_port_ids = list(filter(lambda x: x[1] == attrs["protocol"], obj.item.ports)) for port_id in existing_port_ids: if portInPortRange(new_port_id[0], port_id[0]): # the range is wholly contained already, so just warn _name = obj.item.derived_from_zone if isinstance(obj.item, Policy) else obj.item.name log.warning(FirewallError(errors.ALREADY_ENABLED, "'%s:%s' already in '%s'" % (new_port_id[0], attrs["protocol"], _name))) break # for else: # the range can be coalesced into the existing set added_ranges, removed_ranges = coalescePortRange(new_port_id[0], [_port for (_port, _protocol) in existing_port_ids]) for _range in removed_ranges: entry = (portStr(_range, "-"), attrs["protocol"]) obj.item.ports.remove(entry) for _range in added_ranges: entry = (portStr(_range, "-"), attrs["protocol"]) obj.item.ports.append(entry) elif name == "protocol": if obj._rule: if obj._rule.element: log.warning("Invalid rule: More than one element in rule '%s', ignoring.", str(obj._rule)) obj._rule_error = True return True obj._rule.element = rich.Rich_Protocol(attrs["value"]) else: check_protocol(attrs["value"]) if attrs["value"] not in obj.item.protocols: obj.item.protocols.append(attrs["value"]) else: log.warning("Protocol '%s' already set, ignoring.", attrs["value"]) elif name == "tcp-mss-clamp": if obj._rule: if obj._rule.element: log.warning("Invalid rule: More than one element in rule '%s', ignoring.", str(obj._rule)) obj._rule_error = True return True _value="pmtu" if "value" in attrs and attrs["value"] not in [None, "None"]: _value = attrs["value"] obj._rule.element = rich.Rich_Tcp_Mss_Clamp(_value) else: s = "" if "value" in attrs: s = f" (value='{attrs['value']})'" log.warning("Invalid rule: tcp-mss-clamp%s outside of rule", s) elif name == "icmp-block": if obj._rule: if obj._rule.element: log.warning("Invalid rule: More than one element in rule '%s', ignoring.", str(obj._rule)) obj._rule_error = True return True obj._rule.element = rich.Rich_IcmpBlock(attrs["name"]) return True if attrs["name"] not in obj.item.icmp_blocks: obj.item.icmp_blocks.append(attrs["name"]) else: log.warning("icmp-block '%s' already set, ignoring.", attrs["name"]) elif name == "icmp-type": if obj._rule: if obj._rule.element: log.warning("Invalid rule: More than one element in rule '%s', ignoring.", str(obj._rule)) obj._rule_error = True return True obj._rule.element = rich.Rich_IcmpType(attrs["name"]) return True else: log.warning("Invalid rule: icmp-block '%s' outside of rule", attrs["name"]) elif name == "masquerade": if obj._rule: if obj._rule.element: log.warning("Invalid rule: More than one element in rule '%s', ignoring.", str(obj._rule)) obj._rule_error = True return True obj._rule.element = rich.Rich_Masquerade() else: if obj.item.masquerade: log.warning("Masquerade already set, ignoring.") else: obj.item.masquerade = True elif name == "forward-port": to_port = "" if "to-port" in attrs: to_port = attrs["to-port"] to_addr = "" if "to-addr" in attrs: to_addr = attrs["to-addr"] if obj._rule: if obj._rule.element: log.warning("Invalid rule: More than one element in rule '%s', ignoring.", str(obj._rule)) obj._rule_error = True return True obj._rule.element = rich.Rich_ForwardPort(attrs["port"], attrs["protocol"], to_port, to_addr) return True check_port(attrs["port"]) check_tcpudp(attrs["protocol"]) if to_port: check_port(to_port) if to_addr: if not checkIP(to_addr) and not checkIP6(to_addr): raise FirewallError(errors.INVALID_ADDR, "to-addr '%s' is not a valid address" \ % to_addr) entry = (portStr(attrs["port"], "-"), attrs["protocol"], portStr(to_port, "-"), str(to_addr)) if entry not in obj.item.forward_ports: obj.item.forward_ports.append(entry) else: log.warning("Forward port %s/%s%s%s already set, ignoring.", attrs["port"], attrs["protocol"], " >%s" % to_port if to_port else "", " @%s" % to_addr if to_addr else "") elif name == "source-port": if obj._rule: if obj._rule.element: log.warning("Invalid rule: More than one element in rule '%s', ignoring.", str(obj._rule)) obj._rule_error = True return True obj._rule.element = rich.Rich_SourcePort(attrs["port"], attrs["protocol"]) return True check_port(attrs["port"]) check_tcpudp(attrs["protocol"]) # coalesce and warn about overlapping ranges new_port_id = (portStr(attrs["port"], "-"), attrs["protocol"]) existing_port_ids = list(filter(lambda x: x[1] == attrs["protocol"], obj.item.source_ports)) for port_id in existing_port_ids: if portInPortRange(new_port_id[0], port_id[0]): # the range is wholly contained already, so just warn _name = obj.item.derived_from_zone if isinstance(obj.item, Policy) else obj.item.name log.warning(FirewallError(errors.ALREADY_ENABLED, "'%s:%s' already in '%s'" % (new_port_id[0], attrs["protocol"], _name))) break # for else: # the range can be coalesced into the existing set added_ranges, removed_ranges = coalescePortRange(new_port_id[0], [_port for (_port, _protocol) in existing_port_ids]) for _range in removed_ranges: entry = (portStr(_range, "-"), attrs["protocol"]) obj.item.source_ports.remove(entry) for _range in added_ranges: entry = (portStr(_range, "-"), attrs["protocol"]) obj.item.source_ports.append(entry) elif name == "destination": if not obj._rule: log.warning('Invalid rule: Destination outside of rule') obj._rule_error = True return True if obj._rule.destination: log.warning("Invalid rule: More than one destination in rule '%s', ignoring.", str(obj._rule)) return True invert = False address = None if "address" in attrs: address = attrs["address"] ipset = None if "ipset" in attrs: ipset = attrs["ipset"] if "invert" in attrs and \ attrs["invert"].lower() in [ "yes", "true" ]: invert = True obj._rule.destination = rich.Rich_Destination(address, ipset, invert) elif name in [ "accept", "reject", "drop", "mark" ]: if not obj._rule: log.warning('Invalid rule: Action outside of rule') obj._rule_error = True return True if obj._rule.action: log.warning('Invalid rule: More than one action') obj._rule_error = True return True if name == "accept": obj._rule.action = rich.Rich_Accept() elif name == "reject": _type = None if "type" in attrs: _type = attrs["type"] obj._rule.action = rich.Rich_Reject(_type) elif name == "drop": obj._rule.action = rich.Rich_Drop() elif name == "mark": _set = attrs["set"] obj._rule.action = rich.Rich_Mark(_set) obj._limit_ok = obj._rule.action elif name == "log": if not obj._rule: log.warning('Invalid rule: Log outside of rule') return True if obj._rule.log: log.warning('Invalid rule: More than one log') return True level = None if "level" in attrs: level = attrs["level"] if level not in [ "emerg", "alert", "crit", "error", "warning", "notice", "info", "debug" ]: log.warning('Invalid rule: Invalid log level') obj._rule_error = True return True prefix = None if "prefix" in attrs: prefix = attrs["prefix"] if not prefix or len(prefix) > 127: log.warning('Invalid rule: Invalid log prefix') obj._rule_error = True return True obj._rule.log = rich.Rich_Log(prefix, level) obj._limit_ok = obj._rule.log elif name == "nflog": if not obj._rule: log.warning('Invalid rule: Log outside of rule') return True if obj._rule.log: log.warning('Invalid rule: More than one log') return True group = None if "group" in attrs: group = attrs["group"] if not checkUINT16(group): log.warning('Invalid rule: Invalid nflog group value') obj._rule_error = True return True prefix = None if "prefix" in attrs: prefix = attrs["prefix"] if not prefix or len(prefix) > 127: log.warning('Invalid rule: Invalid nflog prefix') obj._rule_error = True return True threshold = None if "queue-size" in attrs: threshold = attrs["queue-size"] if not checkUINT16(threshold): log.warning('Invalid rule: Invalid nflog queue-size') obj._rule_error = True return True obj._rule.log = rich.Rich_NFLog(group, prefix, threshold) obj._limit_ok = obj._rule.log elif name == "audit": if not obj._rule: log.warning('Invalid rule: Audit outside of rule') return True if obj._rule.audit: log.warning("Invalid rule: More than one audit in rule '%s', ignoring.", str(obj._rule)) obj._rule_error = True return True obj._rule.audit = rich.Rich_Audit() obj._limit_ok = obj._rule.audit elif name == "rule": family = None priority = 0 if "family" in attrs: family = attrs["family"] if family not in [ "ipv4", "ipv6" ]: log.warning('Invalid rule: Rule family "%s" invalid', attrs["family"]) obj._rule_error = True return True if "priority" in attrs: priority = int(attrs["priority"]) obj._rule = rich.Rich_Rule(family=family, priority=priority) elif name == "limit": if not obj._limit_ok: log.warning('Invalid rule: Limit outside of action, log and audit') obj._rule_error = True return True if obj._limit_ok.limit: log.warning("Invalid rule: More than one limit in rule '%s', ignoring.", str(obj._rule)) obj._rule_error = True return True value = attrs["value"] obj._limit_ok.limit = rich.Rich_Limit(value) else: return False return True def common_endElement(obj, name): if name == "rule": if not obj._rule_error: try: obj._rule.check() except Exception as e: log.warning("%s: %s", e, str(obj._rule)) else: if str(obj._rule) not in obj.item.rules_str: obj.item.rules.append(obj._rule) obj.item.rules_str.append(str(obj._rule)) else: log.warning("Rule '%s' already set, ignoring.", str(obj._rule)) obj._rule = None obj._rule_error = False elif name in [ "accept", "reject", "drop", "mark", "log", "audit" ]: obj._limit_ok = None def common_check_config(obj, config, item, all_config, all_io_objects): obj_type = "Policy" if isinstance(obj, Policy) else "Zone" if item == "services" and "services" in all_io_objects: existing_services = all_io_objects["services"] for service in config: if service not in existing_services: raise FirewallError(errors.INVALID_SERVICE, "{} '{}': '{}' not among existing services".format( obj_type, obj.name, service)) elif item == "ports": for port in config: check_port(port[0]) check_tcpudp(port[1]) elif item == "protocols": for proto in config: check_protocol(proto) elif item == "icmp_blocks" and "icmptypes" in all_io_objects: existing_icmptypes = all_io_objects["icmptypes"] for icmptype in config: if icmptype not in existing_icmptypes: ex = FirewallError(errors.INVALID_ICMPTYPE, "{} '{}': '{}' not among existing ICMP types".format( obj_type, obj.name, icmptype)) if icmptype in all_io_objects.get("runtime", {}).get("icmptypes_unsupported", {}): log.debug1("{} (unsupported)".format(ex)) else: raise ex elif item == "forward_ports": for fwd_port in config: check_port(fwd_port[0]) check_tcpudp(fwd_port[1]) if not fwd_port[2] and not fwd_port[3]: raise FirewallError(errors.INVALID_FORWARD, "{} '{}': '{}' is missing to-port AND to-addr ".format( obj_type, obj.name, fwd_port)) if fwd_port[2]: check_port(fwd_port[2]) if fwd_port[3]: if not checkIP(fwd_port[3]) and not checkIP6(fwd_port[3]): raise FirewallError(errors.INVALID_ADDR, "{} '{}': to-addr '{}' is not a valid address".format( obj_type, obj.name, fwd_port[3])) elif item == "source_ports": for port in config: check_port(port[0]) check_tcpudp(port[1]) elif item in ["rules_str", "rich_rules"]: for rule in config: obj_rich = rich.Rich_Rule(rule_str=rule) if obj_rich.element and "icmptypes" in all_io_objects and \ (isinstance(obj_rich.element, rich.Rich_IcmpBlock) or isinstance(obj_rich.element, rich.Rich_IcmpType)): existing_icmptypes = all_io_objects["icmptypes"] if obj_rich.element.name not in existing_icmptypes: ex = FirewallError(errors.INVALID_ICMPTYPE, "{} '{}': '{}' not among existing ICMP types".format( obj_type, obj.name, obj_rich.element.name)) if obj_rich.element.name in all_io_objects.get("runtime", {}).get("icmptypes_unsupported", {}): log.debug1("{} (unsupported)".format(ex)) else: raise ex elif obj_rich.family: ict = all_io_objects["icmptypes"][obj_rich.element.name] if ict.destination and obj_rich.family not in ict.destination: ex = FirewallError(errors.INVALID_ICMPTYPE, "{} '{}': rich rule family '{}' conflicts with icmp type '{}'".format( obj_type, obj.name, obj_rich.family, obj_rich.element.name)) ict_unsupported = all_io_objects.get("runtime", {}).get("icmptypes_unsupported", {}).get(obj_rich.element.name) if ict_unsupported and ict_unsupported.destination and \ obj_rich.family in ict_unsupported.destination: log.debug1("{} (unsupported)".format(ex)) else: raise ex elif isinstance(obj_rich.element, rich.Rich_Service): if obj_rich.element.name not in all_io_objects["services"]: raise FirewallError( errors.INVALID_SERVICE, "{} '{}': '{}' not among existing services".format( obj_type, obj.name, obj_rich.element.name ), ) elif obj_rich.source and obj_rich.source.ipset: if obj_rich.source.ipset not in all_io_objects["ipsets"]: raise FirewallError( errors.INVALID_IPSET, "{} '{}': '{}' not among existing ipsets".format( obj_type, obj.name, obj_rich.source.ipset ), ) elif obj_rich.destination and obj_rich.destination.ipset: if obj_rich.destination.ipset not in all_io_objects["ipsets"]: raise FirewallError( errors.INVALID_IPSET, "{} '{}': '{}' not among existing ipsets".format( obj_type, obj.name, obj_rich.destination.ipset ), ) def common_writer(obj, handler): # short if obj.short and obj.short != "": handler.ignorableWhitespace(" ") handler.startElement("short", { }) handler.characters(obj.short) handler.endElement("short") handler.ignorableWhitespace("\n") # description if obj.description and obj.description != "": handler.ignorableWhitespace(" ") handler.startElement("description", { }) handler.characters(obj.description) handler.endElement("description") handler.ignorableWhitespace("\n") # services for service in uniqify(obj.services): handler.ignorableWhitespace(" ") handler.simpleElement("service", { "name": service }) handler.ignorableWhitespace("\n") # ports for port in uniqify(obj.ports): handler.ignorableWhitespace(" ") handler.simpleElement("port", { "port": port[0], "protocol": port[1] }) handler.ignorableWhitespace("\n") # protocols for protocol in uniqify(obj.protocols): handler.ignorableWhitespace(" ") handler.simpleElement("protocol", { "value": protocol }) handler.ignorableWhitespace("\n") # icmp-blocks for icmp in uniqify(obj.icmp_blocks): handler.ignorableWhitespace(" ") handler.simpleElement("icmp-block", { "name": icmp }) handler.ignorableWhitespace("\n") # masquerade if obj.masquerade: handler.ignorableWhitespace(" ") handler.simpleElement("masquerade", { }) handler.ignorableWhitespace("\n") # forward-ports for forward in uniqify(obj.forward_ports): handler.ignorableWhitespace(" ") attrs = { "port": forward[0], "protocol": forward[1] } if forward[2] and forward[2] != "" : attrs["to-port"] = forward[2] if forward[3] and forward[3] != "" : attrs["to-addr"] = forward[3] handler.simpleElement("forward-port", attrs) handler.ignorableWhitespace("\n") # source-ports for port in uniqify(obj.source_ports): handler.ignorableWhitespace(" ") handler.simpleElement("source-port", { "port": port[0], "protocol": port[1] }) handler.ignorableWhitespace("\n") # rules for rule in obj.rules: attrs = { } if rule.family: attrs["family"] = rule.family if rule.priority != 0: attrs["priority"] = str(rule.priority) handler.ignorableWhitespace(" ") handler.startElement("rule", attrs) handler.ignorableWhitespace("\n") # source if rule.source: attrs = { } if rule.source.addr: attrs["address"] = rule.source.addr if rule.source.mac: attrs["mac"] = rule.source.mac if rule.source.ipset: attrs["ipset"] = rule.source.ipset if rule.source.invert: attrs["invert"] = "True" handler.ignorableWhitespace(" ") handler.simpleElement("source", attrs) handler.ignorableWhitespace("\n") # destination if rule.destination: attrs = { } if rule.destination.addr: attrs["address"] = rule.destination.addr if rule.destination.ipset: attrs["ipset"] = rule.destination.ipset if rule.destination.invert: attrs["invert"] = "True" handler.ignorableWhitespace(" ") handler.simpleElement("destination", attrs) handler.ignorableWhitespace("\n") # element if rule.element: element = "" attrs = { } if type(rule.element) == rich.Rich_Service: element = "service" attrs["name"] = rule.element.name elif type(rule.element) == rich.Rich_Port: element = "port" attrs["port"] = rule.element.port attrs["protocol"] = rule.element.protocol elif type(rule.element) == rich.Rich_Protocol: element = "protocol" attrs["value"] = rule.element.value elif type(rule.element) == rich.Rich_Tcp_Mss_Clamp: element = "tcp-mss-clamp" if rule.element.value and rule.element.value != "pmtu": attrs["value"] = rule.element.value elif type(rule.element) == rich.Rich_Masquerade: element = "masquerade" elif type(rule.element) == rich.Rich_IcmpBlock: element = "icmp-block" attrs["name"] = rule.element.name elif type(rule.element) == rich.Rich_IcmpType: element = "icmp-type" attrs["name"] = rule.element.name elif type(rule.element) == rich.Rich_ForwardPort: element = "forward-port" attrs["port"] = rule.element.port attrs["protocol"] = rule.element.protocol if rule.element.to_port != "": attrs["to-port"] = rule.element.to_port if rule.element.to_address != "": attrs["to-addr"] = rule.element.to_address elif type(rule.element) == rich.Rich_SourcePort: element = "source-port" attrs["port"] = rule.element.port attrs["protocol"] = rule.element.protocol else: raise FirewallError( errors.INVALID_OBJECT, "Unknown element '%s' in obj_writer" % type(rule.element)) handler.ignorableWhitespace(" ") handler.simpleElement(element, attrs) handler.ignorableWhitespace("\n") # rule.element # log if rule.log: if type(rule.log) == rich.Rich_Log: attrs = { } if rule.log.prefix: attrs["prefix"] = rule.log.prefix if rule.log.level: attrs["level"] = rule.log.level if rule.log.limit: handler.ignorableWhitespace(" ") handler.startElement("log", attrs) handler.ignorableWhitespace("\n ") handler.simpleElement("limit", { "value": rule.log.limit.value }) handler.ignorableWhitespace("\n ") handler.endElement("log") else: handler.ignorableWhitespace(" ") handler.simpleElement("log", attrs) handler.ignorableWhitespace("\n") else: attrs = { } if rule.log.group: attrs["group"] = rule.log.group if rule.log.prefix: attrs["prefix"] = rule.log.prefix if rule.log.threshold: attrs["queue-size"] = rule.log.threshold if rule.log.limit: handler.ignorableWhitespace(" ") handler.startElement("nflog", attrs) handler.ignorableWhitespace("\n ") handler.simpleElement("limit", { "value": rule.log.limit.value }) handler.ignorableWhitespace("\n ") handler.endElement("nflog") else: handler.ignorableWhitespace(" ") handler.simpleElement("nflog", attrs) handler.ignorableWhitespace("\n") # audit if rule.audit: attrs = {} if rule.audit.limit: handler.ignorableWhitespace(" ") handler.startElement("audit", { }) handler.ignorableWhitespace("\n ") handler.simpleElement("limit", { "value": rule.audit.limit.value }) handler.ignorableWhitespace("\n ") handler.endElement("audit") else: handler.ignorableWhitespace(" ") handler.simpleElement("audit", attrs) handler.ignorableWhitespace("\n") # action if rule.action: action = "" attrs = { } if type(rule.action) == rich.Rich_Accept: action = "accept" elif type(rule.action) == rich.Rich_Reject: action = "reject" if rule.action.type: attrs["type"] = rule.action.type elif type(rule.action) == rich.Rich_Drop: action = "drop" elif type(rule.action) == rich.Rich_Mark: action = "mark" attrs["set"] = rule.action.set else: log.warning("Unknown action '%s'", type(rule.action)) if rule.action.limit: handler.ignorableWhitespace(" ") handler.startElement(action, attrs) handler.ignorableWhitespace("\n ") handler.simpleElement("limit", { "value": rule.action.limit.value }) handler.ignorableWhitespace("\n ") handler.endElement(action) else: handler.ignorableWhitespace(" ") handler.simpleElement(action, attrs) handler.ignorableWhitespace("\n") handler.ignorableWhitespace(" ") handler.endElement("rule") handler.ignorableWhitespace("\n") class Policy(IO_Object): priority_min = -32768 priority_max = 32767 priority_default = DEFAULT_POLICY_PRIORITY priority_reserved = [0] IMPORT_EXPORT_STRUCTURE = ( ( "version", "" ), # s ( "short", "" ), # s ( "description", "" ), # s ( "target", "" ), # s ( "services", [ "", ], ), # as ( "ports", [ ( "", "" ), ], ), # a(ss) ( "icmp_blocks", [ "", ], ), # as ( "masquerade", False ), # b ( "forward_ports", [ ( "", "", "", "" ), ], ), # a(ssss) ( "rich_rules", [ "" ] ), # as ( "protocols", [ "", ], ), # as ( "source_ports", [ ( "", "" ), ], ), # a(ss) ( "priority", 0 ), # i ( "ingress_zones", [ "" ] ), # as ( "egress_zones", [ "" ] ), # as ) ADDITIONAL_ALNUM_CHARS = [ "_", "-", "/" ] PARSER_REQUIRED_ELEMENT_ATTRS = { "short": None, "description": None, "policy": ["target"], "service": [ "name" ], "port": [ "port", "protocol" ], "icmp-block": [ "name" ], "icmp-type": [ "name" ], "masquerade": None, "forward-port": [ "port", "protocol" ], "rule": None, "source": None, "destination": None, "protocol": [ "value" ], "source-port": [ "port", "protocol" ], "log": None, "nflog": None, "audit": None, "accept": None, "reject": None, "drop": None, "mark": [ "set" ], "limit": [ "value" ], "ingress-zone": [ "name" ], "egress-zone": [ "name" ], } PARSER_OPTIONAL_ELEMENT_ATTRS = { "policy": [ "version", "priority" ], "forward-port": [ "to-port", "to-addr" ], "rule": [ "family", "priority" ], "source": [ "address", "mac", "invert", "family", "ipset" ], "destination": [ "address", "invert", "ipset" ], "log": [ "prefix", "level" ], "nflog": [ "group", "prefix", "queue-size" ], "reject": [ "type" ], "tcp-mss-clamp": [ "value" ], } def __init__(self): super(Policy, self).__init__() self.version = "" self.short = "" self.description = "" self.target = DEFAULT_POLICY_TARGET self.services = [ ] self.ports = [ ] self.protocols = [ ] self.icmp_blocks = [ ] self.icmp_block_inversion = False # for zones, not written to policy config self.masquerade = False self.forward_ports = [ ] self.source_ports = [ ] self.rules = [ ] self.rules_str = [ ] self.applied = False self.priority = self.priority_default self.derived_from_zone = None self.ingress_zones = [] self.egress_zones = [] def cleanup(self): self.version = "" self.short = "" self.description = "" self.target = DEFAULT_POLICY_TARGET del self.services[:] del self.ports[:] del self.protocols[:] del self.icmp_blocks[:] self.icmp_block_inversion = False self.masquerade = False del self.forward_ports[:] del self.source_ports[:] del self.rules[:] del self.rules_str[:] self.applied = False self.priority = self.priority_default del self.ingress_zones[:] del self.egress_zones[:] def __getattr__(self, name): if name == "rich_rules": return self.rules_str else: return getattr(super(Policy, self), name) def __setattr__(self, name, value): if name == "rich_rules": self.rules = [rich.Rich_Rule(rule_str=s) for s in value] # must convert back to string to get the canonical string. self.rules_str = [str(s) for s in self.rules] else: super(Policy, self).__setattr__(name, value) def _check_config(self, config, item, all_config, all_io_objects): common_check_config(self, config, item, all_config, all_io_objects) if self.name in all_io_objects["zones"]: raise FirewallError(errors.NAME_CONFLICT, "Policy '{}': Can't have the same name as a zone.".format(self.name)) if item == "target": if config not in POLICY_TARGETS: raise FirewallError(errors.INVALID_TARGET, "Policy '{}': '{}' is invalid target".format(self.name, config)) elif item == "priority": if config in self.priority_reserved or \ config > self.priority_max or \ config < self.priority_min: raise FirewallError(errors.INVALID_PRIORITY, "Policy '{}': {} is invalid priority. Must be in range [{}, {}]. The following are reserved: {}".format( self.name, config, self.priority_min, self.priority_max, self.priority_reserved)) elif item in ["ingress_zones", "egress_zones"]: existing_zones = ["ANY", "HOST"] + list(all_io_objects["zones"].keys()) for zone in config: if zone not in existing_zones: raise FirewallError(errors.INVALID_ZONE, "Policy '{}': '{}' not among existing zones".format( self.name, zone)) if ((zone not in ["ANY", "HOST"] and (set(["ANY", "HOST"]) & set(config))) or \ (zone in ["ANY", "HOST"] and (set(config) - set([zone])))): raise FirewallError(errors.INVALID_ZONE, "Policy '{}': '{}' may only contain one of: many regular zones, ANY, or HOST".format( self.name, item)) if zone == "HOST" and \ ((item == "ingress_zones" and "egress_zones" in all_config and "HOST" in all_config["egress_zones"]) or \ (item == "egress_zones" and "ingress_zones" in all_config and "HOST" in all_config["ingress_zones"])): raise FirewallError(errors.INVALID_ZONE, "Policy '{}': 'HOST' can only appear in either ingress or egress zones, but not both".format( self.name)) elif item == "masquerade" and config: if "egress_zones" in all_config and "HOST" in all_config["egress_zones"]: raise FirewallError(errors.INVALID_ZONE, "Policy '{}': 'masquerade' is invalid for egress zone 'HOST'".format( self.name)) elif "ingress_zones" in all_config: if "HOST" in all_config["ingress_zones"]: raise FirewallError(errors.INVALID_ZONE, "Policy '{}': 'masquerade' is invalid for ingress zone 'HOST'".format( self.name)) for zone in all_config["ingress_zones"]: if zone == "ANY": continue if zone not in all_io_objects["zones"]: raise FirewallError(errors.INVALID_ZONE, "Policy '{}': Zone '{}' does not exist.".format(self.name, zone)) if all_io_objects["conf"].get("FirewallBackend") != "nftables" \ and all_io_objects["zones"][zone].interfaces: raise FirewallError(errors.INVALID_ZONE, "Policy '{}': 'masquerade' cannot be used because ingress zone '{}' has assigned interfaces. ".format(self.name, zone)) elif item == "rich_rules": for rule in config: obj = rich.Rich_Rule(rule_str=rule) if obj.element and isinstance(obj.element, rich.Rich_Masquerade): if "egress_zones" in all_config and "HOST" in all_config["egress_zones"]: raise FirewallError(errors.INVALID_ZONE, "Policy '{}': 'masquerade' is invalid for egress zone 'HOST'".format( self.name)) elif "ingress_zones" in all_config: if "HOST" in all_config["ingress_zones"]: raise FirewallError(errors.INVALID_ZONE, "Policy '{}': 'masquerade' is invalid for ingress zone 'HOST'".format( self.name)) for zone in all_config["ingress_zones"]: if zone == "ANY": continue if zone not in all_io_objects["zones"]: raise FirewallError(errors.INVALID_ZONE, "Policy '{}': Zone '{}' does not exist.".format(self.name, zone)) if all_io_objects["conf"].get("FirewallBackend") != "nftables" \ and all_io_objects["zones"][zone].interfaces: raise FirewallError(errors.INVALID_ZONE, "Policy '{}': 'masquerade' cannot be used because ingress zone '{}' has assigned interfaces. ".format(self.name, zone)) elif obj.element and isinstance(obj.element, rich.Rich_ForwardPort): if "egress_zones" in all_config: if "HOST" in all_config["egress_zones"]: if obj.element.to_address: raise FirewallError(errors.INVALID_FORWARD, "Policy '{}': A 'forward-port' with 'to-addr' is invalid for egress zone 'HOST'".format( self.name)) elif all_config["egress_zones"]: if not obj.element.to_address: raise FirewallError(errors.INVALID_FORWARD, "Policy '{}': 'forward-port' requires 'to-addr' if egress zone is 'ANY' or a zone".format( self.name)) if "ANY" not in all_config["egress_zones"]: for zone in all_config["egress_zones"]: if zone not in all_io_objects["zones"]: raise FirewallError(errors.INVALID_ZONE, "Policy '{}': Zone '{}' does not exist.".format(self.name, zone)) if all_io_objects["zones"][zone].interfaces: raise FirewallError(errors.INVALID_ZONE, "Policy '{}': 'forward-port' cannot be used because egress zone '{}' has assigned interfaces".format(self.name, zone)) elif obj.action and isinstance(obj.action, rich.Rich_Mark): if "egress_zones" in all_config: for zone in all_config["egress_zones"]: if zone in ["ANY", "HOST"]: continue if zone not in all_io_objects["zones"]: raise FirewallError(errors.INVALID_ZONE, "Policy '{}': Zone '{}' does not exist.".format(self.name, zone)) if all_io_objects["zones"][zone].interfaces: raise FirewallError(errors.INVALID_ZONE, "Policy '{}': 'mark' action cannot be used because egress zone '{}' has assigned interfaces".format(self.name, zone)) elif item == "forward_ports": for fwd_port in config: if "egress_zones" in all_config: if "HOST" in all_config["egress_zones"]: if fwd_port[3]: raise FirewallError(errors.INVALID_FORWARD, "Policy '{}': A 'forward-port' with 'to-addr' is invalid for egress zone 'HOST'".format( self.name)) elif all_config["egress_zones"]: if not fwd_port[3]: raise FirewallError(errors.INVALID_FORWARD, "Policy '{}': 'forward-port' requires 'to-addr' if egress zone is 'ANY' or a zone".format( self.name)) if "ANY" not in all_config["egress_zones"]: for zone in all_config["egress_zones"]: if zone not in all_io_objects["zones"]: raise FirewallError(errors.INVALID_ZONE, "Policy '{}': Zone '{}' does not exist.".format(self.name, zone)) if all_io_objects["zones"][zone].interfaces: raise FirewallError(errors.INVALID_ZONE, "Policy '{}': 'forward-port' cannot be used because egress zone '{}' has assigned interfaces".format(self.name, zone)) def check_name(self, name): super(Policy, self).check_name(name) if name.startswith('/'): raise FirewallError(errors.INVALID_NAME, "Policy '{}': name can't start with '/'".format(name)) elif name.endswith('/'): raise FirewallError(errors.INVALID_NAME, "Policy '{}': name can't end with '/'".format(name)) elif name.count('/') > 1: raise FirewallError(errors.INVALID_NAME, "Policy '{}': name has more than one '/'".format(name)) else: if "/" in name: checked_name = name[:name.find('/')] else: checked_name = name if len(checked_name) > max_policy_name_len(): raise FirewallError(errors.INVALID_NAME, "Policy '{}': name has {} chars, max is {}".format( name, len(checked_name), max_policy_name_len())) # PARSER class policy_ContentHandler(IO_Object_ContentHandler): def __init__(self, item): IO_Object_ContentHandler.__init__(self, item) self._rule = None self._rule_error = False self._limit_ok = None def startElement(self, name, attrs): IO_Object_ContentHandler.startElement(self, name, attrs) if self._rule_error: return self.item.parser_check_element_attrs(name, attrs) if common_startElement(self, name, attrs): return elif name == "policy": if "version" in attrs: self.item.version = attrs["version"] if "priority" in attrs: self.item.priority = int(attrs["priority"]) if "target" in attrs: target = attrs["target"] if target not in POLICY_TARGETS: raise FirewallError(errors.INVALID_TARGET, target) if target: self.item.target = target elif name == "ingress-zone": if attrs["name"] not in self.item.ingress_zones: self.item.ingress_zones.append(attrs["name"]) else: log.warning("Ingress zone '%s' already set, ignoring.", attrs["name"]) elif name == "egress-zone": if attrs["name"] not in self.item.egress_zones: self.item.egress_zones.append(attrs["name"]) else: log.warning("Egress zone '%s' already set, ignoring.", attrs["name"]) elif name == "source": if not self._rule: log.warning('Invalid rule: Source outside of rule') self._rule_error = True return if self._rule.source: log.warning("Invalid rule: More than one source in rule '%s', ignoring.", str(self._rule)) self._rule_error = True return invert = False if "invert" in attrs and \ attrs["invert"].lower() in [ "yes", "true" ]: invert = True addr = mac = ipset = None if "address" in attrs: addr = attrs["address"] if "mac" in attrs: mac = attrs["mac"] if "ipset" in attrs: ipset = attrs["ipset"] self._rule.source = rich.Rich_Source(addr, mac, ipset, invert=invert) return else: log.warning("Unknown XML element '%s'", name) return def endElement(self, name): IO_Object_ContentHandler.endElement(self, name) common_endElement(self, name) def policy_reader(filename, path, no_check_name=False): policy = Policy() if not filename.endswith(".xml"): raise FirewallError(errors.INVALID_NAME, "'%s' is missing .xml suffix" % filename) policy.name = filename[:-4] if not no_check_name: policy.check_name(policy.name) policy.filename = filename policy.path = path policy.builtin = False if path.startswith(config.ETC_FIREWALLD) else True policy.default = policy.builtin handler = policy_ContentHandler(policy) parser = sax.make_parser() parser.setContentHandler(handler) name = "%s/%s" % (path, filename) with open(name, "rb") as f: source = sax.InputSource(None) source.setByteStream(f) try: parser.parse(source) except sax.SAXParseException as msg: raise FirewallError(errors.INVALID_POLICY, "not a valid policy file: %s" % \ msg.getException()) del handler del parser return policy def policy_writer(policy, path=None): _path = path if path else policy.path if policy.filename: name = "%s/%s" % (_path, policy.filename) else: name = "%s/%s.xml" % (_path, policy.name) if os.path.exists(name): try: shutil.copy2(name, "%s.old" % name) except Exception as msg: log.error("Backup of file '%s' failed: %s", name, msg) dirpath = os.path.dirname(name) if dirpath.startswith(config.ETC_FIREWALLD) and not os.path.exists(dirpath): if not os.path.exists(config.ETC_FIREWALLD): os.mkdir(config.ETC_FIREWALLD, 0o750) os.mkdir(dirpath, 0o750) f = io.open(name, mode='wt', encoding='UTF-8') handler = IO_Object_XMLGenerator(f) handler.startDocument() # start policy element attrs = {} if policy.version and policy.version != "": attrs["version"] = policy.version if policy.priority != policy.priority_default: attrs["priority"] = str(policy.priority) attrs["target"] = policy.target handler.startElement("policy", attrs) handler.ignorableWhitespace("\n") common_writer(policy, handler) # ingress-zones for zone in uniqify(policy.ingress_zones): handler.ignorableWhitespace(" ") handler.simpleElement("ingress-zone", { "name": zone }) handler.ignorableWhitespace("\n") # egress-zones for zone in uniqify(policy.egress_zones): handler.ignorableWhitespace(" ") handler.simpleElement("egress-zone", { "name": zone }) handler.ignorableWhitespace("\n") # end policy element handler.endElement("policy") handler.ignorableWhitespace("\n") handler.endDocument() f.close() del handler	49.27 KB	2026-03-10 23:36
📄 service.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__ = [ "Service", "service_reader", "service_writer" ] import xml.sax as sax import os import io import shutil from firewall import config from firewall.core.io.io_object import IO_Object, \ IO_Object_ContentHandler, IO_Object_XMLGenerator, check_port, \ check_tcpudp, check_protocol, check_address from firewall.core.logger import log from firewall import errors from firewall.errors import FirewallError class Service(IO_Object): IMPORT_EXPORT_STRUCTURE = ( ( "version", "" ), ( "short", "" ), ( "description", "" ), ( "ports", [ ( "", "" ), ], ), ( "modules", [ "", ], ), ( "destination", { "": "", }, ), ( "protocols", [ "", ], ), ( "source_ports", [ ( "", "" ), ], ), ( "includes", [ "" ], ), ( "helpers", [ "", ], ), ) ADDITIONAL_ALNUM_CHARS = [ "_", "-" ] PARSER_REQUIRED_ELEMENT_ATTRS = { "short": None, "description": None, "service": None, } PARSER_OPTIONAL_ELEMENT_ATTRS = { "service": [ "name", "version" ], "port": [ "port", "protocol" ], "protocol": [ "value" ], "module": [ "name" ], "destination": [ "ipv4", "ipv6" ], "source-port": [ "port", "protocol" ], "include": [ "service" ], "helper": [ "name" ], } def __init__(self): super(Service, self).__init__() self.version = "" self.short = "" self.description = "" self.ports = [ ] self.protocols = [ ] self.modules = [ ] self.destination = { } self.source_ports = [ ] self.includes = [ ] self.helpers = [ ] def cleanup(self): self.version = "" self.short = "" self.description = "" del self.ports[:] del self.protocols[:] del self.modules[:] self.destination.clear() del self.source_ports[:] del self.includes[:] del self.helpers[:] def _check_config(self, config, item, all_config, all_io_objects): if item == "ports": for port in config: if port[0] != "": check_port(port[0]) check_tcpudp(port[1]) else: # only protocol check_protocol(port[1]) elif item == "protocols": for proto in config: check_protocol(proto) elif item == "source_ports": for port in config: check_port(port[0]) check_tcpudp(port[1]) elif item == "destination": for destination in config: if destination not in [ "ipv4", "ipv6" ]: raise FirewallError(errors.INVALID_DESTINATION, "'%s' not in {'ipv4'\|'ipv6'}" % \ destination) check_address(destination, config[destination]) elif item == "modules": for module in config: if module.startswith("nf_conntrack_"): module = module.replace("nf_conntrack_", "") if "_" in module: module = module.replace("_", "-") if len(module) < 2: raise FirewallError(errors.INVALID_MODULE, module) elif item == "includes": for include in config: if include not in all_io_objects["services"]: raise FirewallError(errors.INVALID_SERVICE, "Service '{}': Included service '{}' not found.".format( self.name, include)) # PARSER class service_ContentHandler(IO_Object_ContentHandler): def startElement(self, name, attrs): IO_Object_ContentHandler.startElement(self, name, attrs) self.item.parser_check_element_attrs(name, attrs) if name == "service": if "name" in attrs: log.warning("Ignoring deprecated attribute name='%s'", attrs["name"]) if "version" in attrs: self.item.version = attrs["version"] elif name == "short": pass elif name == "description": pass elif name == "port": if attrs["port"] != "": check_port(attrs["port"]) check_tcpudp(attrs["protocol"]) entry = (attrs["port"], attrs["protocol"]) if entry not in self.item.ports: self.item.ports.append(entry) else: log.warning("Port '%s/%s' already set, ignoring.", attrs["port"], attrs["protocol"]) else: check_protocol(attrs["protocol"]) if attrs["protocol"] not in self.item.protocols: self.item.protocols.append(attrs["protocol"]) else: log.warning("Protocol '%s' already set, ignoring.", attrs["protocol"]) elif name == "protocol": check_protocol(attrs["value"]) if attrs["value"] not in self.item.protocols: self.item.protocols.append(attrs["value"]) else: log.warning("Protocol '%s' already set, ignoring.", attrs["value"]) elif name == "source-port": check_port(attrs["port"]) check_tcpudp(attrs["protocol"]) entry = (attrs["port"], attrs["protocol"]) if entry not in self.item.source_ports: self.item.source_ports.append(entry) else: log.warning("SourcePort '%s/%s' already set, ignoring.", attrs["port"], attrs["protocol"]) elif name == "destination": for x in [ "ipv4", "ipv6" ]: if x in attrs: check_address(x, attrs[x]) if x in self.item.destination: log.warning("Destination address for '%s' already set, ignoring", x) else: self.item.destination[x] = attrs[x] elif name == "module": module = attrs["name"] if module.startswith("nf_conntrack_"): module = module.replace("nf_conntrack_", "") if "_" in module: module = module.replace("_", "-") if module not in self.item.modules: self.item.modules.append(module) else: log.warning("Module '%s' already set, ignoring.", module) elif name == "include": if attrs["service"] not in self.item.includes: self.item.includes.append(attrs["service"]) else: log.warning("Include '%s' already set, ignoring.", attrs["service"]) elif name == "helper": if attrs["name"] not in self.item.helpers: self.item.helpers.append(attrs["name"]) else: log.warning("Helper '%s' already set, ignoring.", attrs["name"]) def service_reader(filename, path): service = Service() if not filename.endswith(".xml"): raise FirewallError(errors.INVALID_NAME, "'%s' is missing .xml suffix" % filename) service.name = filename[:-4] service.check_name(service.name) service.filename = filename service.path = path service.builtin = False if path.startswith(config.ETC_FIREWALLD) else True service.default = service.builtin handler = service_ContentHandler(service) parser = sax.make_parser() parser.setContentHandler(handler) name = "%s/%s" % (path, filename) with open(name, "rb") as f: source = sax.InputSource(None) source.setByteStream(f) try: parser.parse(source) except sax.SAXParseException as msg: raise FirewallError(errors.INVALID_SERVICE, "not a valid service file: %s" % \ msg.getException()) del handler del parser return service def service_writer(service, path=None): _path = path if path else service.path if service.filename: name = "%s/%s" % (_path, service.filename) else: name = "%s/%s.xml" % (_path, service.name) if os.path.exists(name): try: shutil.copy2(name, "%s.old" % name) except Exception as msg: log.error("Backup of file '%s' failed: %s", name, msg) dirpath = os.path.dirname(name) if dirpath.startswith(config.ETC_FIREWALLD) and not os.path.exists(dirpath): if not os.path.exists(config.ETC_FIREWALLD): os.mkdir(config.ETC_FIREWALLD, 0o750) os.mkdir(dirpath, 0o750) f = io.open(name, mode='wt', encoding='UTF-8') handler = IO_Object_XMLGenerator(f) handler.startDocument() # start service element attrs = {} if service.version and service.version != "": attrs["version"] = service.version handler.startElement("service", attrs) handler.ignorableWhitespace("\n") # short if service.short and service.short != "": handler.ignorableWhitespace(" ") handler.startElement("short", { }) handler.characters(service.short) handler.endElement("short") handler.ignorableWhitespace("\n") # description if service.description and service.description != "": handler.ignorableWhitespace(" ") handler.startElement("description", { }) handler.characters(service.description) handler.endElement("description") handler.ignorableWhitespace("\n") # ports for port in service.ports: handler.ignorableWhitespace(" ") handler.simpleElement("port", { "port": port[0], "protocol": port[1] }) handler.ignorableWhitespace("\n") # protocols for protocol in service.protocols: handler.ignorableWhitespace(" ") handler.simpleElement("protocol", { "value": protocol }) handler.ignorableWhitespace("\n") # source ports for port in service.source_ports: handler.ignorableWhitespace(" ") handler.simpleElement("source-port", { "port": port[0], "protocol": port[1] }) handler.ignorableWhitespace("\n") # modules for module in service.modules: handler.ignorableWhitespace(" ") handler.simpleElement("module", { "name": module }) handler.ignorableWhitespace("\n") # destination if len(service.destination) > 0: handler.ignorableWhitespace(" ") handler.simpleElement("destination", service.destination) handler.ignorableWhitespace("\n") # includes for include in service.includes: handler.ignorableWhitespace(" ") handler.simpleElement("include", { "service": include }) handler.ignorableWhitespace("\n") # helpers for helper in service.helpers: handler.ignorableWhitespace(" ") handler.simpleElement("helper", { "name": helper }) handler.ignorableWhitespace("\n") # end service element handler.endElement('service') handler.ignorableWhitespace("\n") handler.endDocument() f.close() del handler	12.04 KB	2026-03-10 23:36
📄 zone.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__ = [ "Zone", "zone_reader", "zone_writer" ] import xml.sax as sax import os import io import shutil from firewall import config from firewall.functions import checkIPnMask, checkIP6nMask, checkInterface, uniqify, max_zone_name_len, check_mac from firewall.core.base import DEFAULT_ZONE_TARGET, ZONE_TARGETS from firewall.core.io.io_object import IO_Object, \ IO_Object_ContentHandler, IO_Object_XMLGenerator from firewall.core.io.policy import common_startElement, common_endElement, common_check_config, common_writer from firewall.core import rich from firewall.core.logger import log from firewall import errors from firewall.errors import FirewallError class Zone(IO_Object): """ Zone class """ IMPORT_EXPORT_STRUCTURE = ( ( "version", "" ), # s ( "short", "" ), # s ( "description", "" ), # s ( "UNUSED", False ), # b ( "target", "" ), # s ( "services", [ "", ], ), # as ( "ports", [ ( "", "" ), ], ), # a(ss) ( "icmp_blocks", [ "", ], ), # as ( "masquerade", False ), # b ( "forward_ports", [ ( "", "", "", "" ), ], ), # a(ssss) ( "interfaces", [ "" ] ), # as ( "sources", [ "" ] ), # as ( "rules_str", [ "" ] ), # as ( "protocols", [ "", ], ), # as ( "source_ports", [ ( "", "" ), ], ), # a(ss) ( "icmp_block_inversion", False ), # b ( "forward", True ), # b ) ADDITIONAL_ALNUM_CHARS = [ "_", "-", "/" ] PARSER_REQUIRED_ELEMENT_ATTRS = { "short": None, "description": None, "zone": None, "service": [ "name" ], "port": [ "port", "protocol" ], "icmp-block": [ "name" ], "icmp-type": [ "name" ], "forward": None, "forward-port": [ "port", "protocol" ], "interface": [ "name" ], "rule": None, "source": None, "destination": None, "protocol": [ "value" ], "source-port": [ "port", "protocol" ], "log": None, "nflog": None, "audit": None, "accept": None, "reject": None, "drop": None, "mark": [ "set" ], "limit": [ "value" ], "icmp-block-inversion": None, } PARSER_OPTIONAL_ELEMENT_ATTRS = { "zone": [ "name", "immutable", "target", "version" ], "masquerade": [ "enabled" ], "forward-port": [ "to-port", "to-addr" ], "rule": [ "family", "priority" ], "source": [ "address", "mac", "invert", "family", "ipset" ], "destination": [ "address", "invert", "ipset" ], "log": [ "prefix", "level" ], "nflog": [ "group", "prefix", "queue-size" ], "reject": [ "type" ], "tcp-mss-clamp": [ "value" ], } @staticmethod def index_of(element): for i, (el, dummy) in enumerate(Zone.IMPORT_EXPORT_STRUCTURE): if el == element: return i raise FirewallError(errors.UNKNOWN_ERROR, "index_of()") def __init__(self): super(Zone, self).__init__() self.version = "" self.short = "" self.description = "" self.UNUSED = False self.target = DEFAULT_ZONE_TARGET self.services = [ ] self.ports = [ ] self.protocols = [ ] self.icmp_blocks = [ ] self.forward = True self.masquerade = False self.forward_ports = [ ] self.source_ports = [ ] self.interfaces = [ ] self.sources = [ ] self.rules = [ ] self.rules_str = [ ] self.icmp_block_inversion = False self.combined = False self.applied = False def cleanup(self): self.version = "" self.short = "" self.description = "" self.UNUSED = False self.target = DEFAULT_ZONE_TARGET del self.services[:] del self.ports[:] del self.protocols[:] del self.icmp_blocks[:] self.forward = True self.masquerade = False del self.forward_ports[:] del self.source_ports[:] del self.interfaces[:] del self.sources[:] del self.rules[:] del self.rules_str[:] self.icmp_block_inversion = False self.combined = False self.applied = False def __setattr__(self, name, value): if name == "rules_str": self.rules = [rich.Rich_Rule(rule_str=s) for s in value] # must convert back to string to get the canonical string. super(Zone, self).__setattr__(name, [str(s) for s in self.rules]) else: super(Zone, self).__setattr__(name, value) def export_config_dict(self): conf = super(Zone, self).export_config_dict() del conf["UNUSED"] return conf def _check_config(self, config, item, all_config, all_io_objects): common_check_config(self, config, item, all_config, all_io_objects) if self.name in all_io_objects["policies"]: raise FirewallError(errors.NAME_CONFLICT, "Zone '{}': Can't have the same name as a policy.".format(self.name)) if item == "target": if config not in ZONE_TARGETS: raise FirewallError(errors.INVALID_TARGET, "Zone '{}': invalid target '{}'".format( self.name, config)) elif item == "interfaces": for interface in config: if not checkInterface(interface): raise FirewallError(errors.INVALID_INTERFACE, "Zone '{}': invalid interface '{}'".format( self.name, interface)) for zone in all_io_objects["zones"]: if zone == self.name: continue if interface in all_io_objects["zones"][zone].interfaces: raise FirewallError(errors.INVALID_INTERFACE, "Zone '{}': interface '{}' already bound to zone '{}'".format( self.name, interface, zone)) elif item == "sources": for source in config: if not checkIPnMask(source) and not checkIP6nMask(source) and \ not check_mac(source) and not source.startswith("ipset:"): raise FirewallError(errors.INVALID_ADDR, "Zone '{}': invalid source '{}'".format( self.name, source)) for zone in all_io_objects["zones"]: if zone == self.name: continue if source in all_io_objects["zones"][zone].sources: raise FirewallError(errors.INVALID_ADDR, "Zone '{}': source '{}' already bound to zone '{}'".format( self.name, source, zone)) def check_name(self, name): super(Zone, self).check_name(name) if name.startswith('/'): raise FirewallError(errors.INVALID_NAME, "Zone '{}': name can't start with '/'".format(name)) elif name.endswith('/'): raise FirewallError(errors.INVALID_NAME, "Zone '{}': name can't end with '/'".format(name)) elif name.count('/') > 1: raise FirewallError(errors.INVALID_NAME, "Zone '{}': name has more than one '/'".format(name)) else: if "/" in name: checked_name = name[:name.find('/')] else: checked_name = name if len(checked_name) > max_zone_name_len(): raise FirewallError(errors.INVALID_NAME, "Zone '{}': name has {} chars, max is {}".format( name, len(checked_name), max_zone_name_len())) def combine(self, zone): self.combined = True self.filename = None self.version = "" self.short = "" self.description = "" for interface in zone.interfaces: if interface not in self.interfaces: self.interfaces.append(interface) for source in zone.sources: if source not in self.sources: self.sources.append(source) for service in zone.services: if service not in self.services: self.services.append(service) for port in zone.ports: if port not in self.ports: self.ports.append(port) for proto in zone.protocols: if proto not in self.protocols: self.protocols.append(proto) for icmp in zone.icmp_blocks: if icmp not in self.icmp_blocks: self.icmp_blocks.append(icmp) if zone.forward: self.forward = True if zone.masquerade: self.masquerade = True for forward in zone.forward_ports: if forward not in self.forward_ports: self.forward_ports.append(forward) for port in zone.source_ports: if port not in self.source_ports: self.source_ports.append(port) for rule in zone.rules: self.rules.append(rule) self.rules_str.append(str(rule)) if zone.icmp_block_inversion: self.icmp_block_inversion = True # PARSER class zone_ContentHandler(IO_Object_ContentHandler): def __init__(self, item): IO_Object_ContentHandler.__init__(self, item) self._rule = None self._rule_error = False self._limit_ok = None def startElement(self, name, attrs): IO_Object_ContentHandler.startElement(self, name, attrs) if self._rule_error: return self.item.parser_check_element_attrs(name, attrs) if common_startElement(self, name, attrs): return elif name == "zone": if "name" in attrs: log.warning("Ignoring deprecated attribute name='%s'", attrs["name"]) if "version" in attrs: self.item.version = attrs["version"] if "immutable" in attrs: log.warning("Ignoring deprecated attribute immutable='%s'", attrs["immutable"]) if "target" in attrs: target = attrs["target"] if target not in ZONE_TARGETS: raise FirewallError(errors.INVALID_TARGET, target) if target != "" and target != DEFAULT_ZONE_TARGET: self.item.target = target elif name == "forward": if self.item.forward: log.warning("Forward already set, ignoring.") else: self.item.forward = True elif name == "interface": if self._rule: log.warning('Invalid rule: interface use in rule.') self._rule_error = True return # zone bound to interface if "name" not in attrs: log.warning('Invalid interface: Name missing.') self._rule_error = True return if attrs["name"] not in self.item.interfaces: self.item.interfaces.append(attrs["name"]) else: log.warning("Interface '%s' already set, ignoring.", attrs["name"]) elif name == "source": if self._rule: if self._rule.source: log.warning("Invalid rule: More than one source in rule '%s', ignoring.", str(self._rule)) self._rule_error = True return invert = False if "invert" in attrs and \ attrs["invert"].lower() in [ "yes", "true" ]: invert = True addr = mac = ipset = None if "address" in attrs: addr = attrs["address"] if "mac" in attrs: mac = attrs["mac"] if "ipset" in attrs: ipset = attrs["ipset"] self._rule.source = rich.Rich_Source(addr, mac, ipset, invert=invert) return # zone bound to source if "address" not in attrs and "ipset" not in attrs: log.warning('Invalid source: No address no ipset.') return if "address" in attrs and "ipset" in attrs: log.warning('Invalid source: Address and ipset.') return if "family" in attrs: log.warning("Ignoring deprecated attribute family='%s'", attrs["family"]) if "invert" in attrs: log.warning('Invalid source: Invertion not allowed here.') return if "address" in attrs: if not checkIPnMask(attrs["address"]) and \ not checkIP6nMask(attrs["address"]) and \ not check_mac(attrs["address"]): raise FirewallError(errors.INVALID_ADDR, attrs["address"]) if "ipset" in attrs: entry = "ipset:%s" % attrs["ipset"] if entry not in self.item.sources: self.item.sources.append(entry) else: log.warning("Source '%s' already set, ignoring.", attrs["address"]) if "address" in attrs: entry = attrs["address"] if entry not in self.item.sources: self.item.sources.append(entry) else: log.warning("Source '%s' already set, ignoring.", attrs["address"]) elif name == "icmp-block-inversion": if self.item.icmp_block_inversion: log.warning("Icmp-Block-Inversion already set, ignoring.") else: self.item.icmp_block_inversion = True else: log.warning("Unknown XML element '%s'", name) return def endElement(self, name): IO_Object_ContentHandler.endElement(self, name) common_endElement(self, name) def zone_reader(filename, path, no_check_name=False): zone = Zone() if not filename.endswith(".xml"): raise FirewallError(errors.INVALID_NAME, "'%s' is missing .xml suffix" % filename) zone.name = filename[:-4] if not no_check_name: zone.check_name(zone.name) zone.filename = filename zone.path = path zone.builtin = False if path.startswith(config.ETC_FIREWALLD) else True zone.default = zone.builtin # new Zone() objects default this to True, but if reading on disk # configuration we have to assume False, because the absence of # <forward> element indicates False. Presence indicates True. zone.forward = False handler = zone_ContentHandler(zone) parser = sax.make_parser() parser.setContentHandler(handler) name = "%s/%s" % (path, filename) with open(name, "rb") as f: source = sax.InputSource(None) source.setByteStream(f) try: parser.parse(source) except sax.SAXParseException as msg: raise FirewallError(errors.INVALID_ZONE, "not a valid zone file: %s" % \ msg.getException()) del handler del parser return zone def zone_writer(zone, path=None): _path = path if path else zone.path if zone.filename: name = "%s/%s" % (_path, zone.filename) else: name = "%s/%s.xml" % (_path, zone.name) if os.path.exists(name): try: shutil.copy2(name, "%s.old" % name) except Exception as msg: log.error("Backup of file '%s' failed: %s", name, msg) dirpath = os.path.dirname(name) if dirpath.startswith(config.ETC_FIREWALLD) and not os.path.exists(dirpath): if not os.path.exists(config.ETC_FIREWALLD): os.mkdir(config.ETC_FIREWALLD, 0o750) os.mkdir(dirpath, 0o750) f = io.open(name, mode='wt', encoding='UTF-8') handler = IO_Object_XMLGenerator(f) handler.startDocument() # start zone element attrs = {} if zone.version and zone.version != "": attrs["version"] = zone.version if zone.target != DEFAULT_ZONE_TARGET: attrs["target"] = zone.target handler.startElement("zone", attrs) handler.ignorableWhitespace("\n") common_writer(zone, handler) # interfaces for interface in uniqify(zone.interfaces): handler.ignorableWhitespace(" ") handler.simpleElement("interface", { "name": interface }) handler.ignorableWhitespace("\n") # source for source in uniqify(zone.sources): handler.ignorableWhitespace(" ") if "ipset:" in source: handler.simpleElement("source", { "ipset": source[6:] }) else: handler.simpleElement("source", { "address": source }) handler.ignorableWhitespace("\n") # icmp-block-inversion if zone.icmp_block_inversion: handler.ignorableWhitespace(" ") handler.simpleElement("icmp-block-inversion", { }) handler.ignorableWhitespace("\n") # forward if zone.forward: handler.ignorableWhitespace(" ") handler.simpleElement("forward", { }) handler.ignorableWhitespace("\n") # end zone element handler.endElement("zone") handler.ignorableWhitespace("\n") handler.endDocument() f.close() del handler	18.46 KB	2026-03-10 23:36

Terminal