D7net
Home
Console
Upload
information
Create File
Create Folder
About
Tools
:
/
lib
/
python3
/
dist-packages
/
ufw
/
Filename :
common.py
back
Copy
'''common.py: common classes for ufw''' # # Copyright 2008-2018 Canonical Ltd. # # This program is free software: you can redistribute it and/or modify # it under the terms of the GNU General Public License version 3, # as published by the Free Software Foundation. # # 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 re import socket import ufw.util from ufw.util import debug programName = "ufw" state_dir = "/lib/ufw" share_dir = "/usr/share/ufw" trans_dir = share_dir config_dir = "/etc" prefix_dir = "/usr" iptables_dir = "/usr/sbin" do_checks = True class UFWError(Exception): '''This class represents ufw exceptions''' def __init__(self, value): self.value = value def __str__(self): return repr(self.value) class UFWRule: '''This class represents firewall rules''' def __init__(self, action, protocol, dport="any", dst="0.0.0.0/0", sport="any", src="0.0.0.0/0", direction="in", forward=False, comment=""): # Be sure to update dup_rule accordingly... self.remove = False self.updated = False self.v6 = False self.dst = "" self.src = "" self.dport = "" self.sport = "" self.protocol = "" self.multi = False self.dapp = "" self.sapp = "" self.action = "" self.position = 0 self.logtype = "" self.interface_in = "" self.interface_out = "" self.direction = "" self.forward = forward self.comment = "" try: self.set_action(action) self.set_protocol(protocol) self.set_port(dport) self.set_port(sport, "src") self.set_src(src) self.set_dst(dst) self.set_direction(direction) self.set_comment(comment) except UFWError: raise def __str__(self): return self.format_rule() def _get_attrib(self): '''Print rule to stdout''' res = "'%s'" % (self) keys = list(self.__dict__) keys.sort() for k in keys: res += ", %s=%s" % (k, self.__dict__[k]) return res def dup_rule(self): '''Return a duplicate of a rule''' rule = UFWRule(self.action, self.protocol) rule.remove = self.remove rule.updated = self.updated rule.v6 = self.v6 rule.dst = self.dst rule.src = self.src rule.dport = self.dport rule.sport = self.sport rule.multi = self.multi rule.dapp = self.dapp rule.sapp = self.sapp rule.position = self.position rule.logtype = self.logtype rule.interface_in = self.interface_in rule.interface_out = self.interface_out rule.direction = self.direction rule.forward = self.forward rule.comment = self.comment return rule def format_rule(self): '''Format rule for later parsing''' rule_str = "" if self.interface_in != "": rule_str += " -i %s" % (self.interface_in) if self.interface_out != "": rule_str += " -o %s" % (self.interface_out) # Protocol is handled below if self.protocol == "any": rule_str += " -p all" else: rule_str += " -p " + self.protocol if self.multi: rule_str += " -m multiport" if self.dport != "any" and self.sport != "any": rule_str += " --dports " + self.dport rule_str += " -m multiport" rule_str += " --sports " + self.sport elif self.dport != "any": rule_str += " --dports " + self.dport elif self.sport != "any": rule_str += " --sports " + self.sport if self.dst != "0.0.0.0/0" and self.dst != "::/0": rule_str += " -d " + self.dst if not self.multi and self.dport != "any": rule_str += " --dport " + self.dport if self.src != "0.0.0.0/0" and self.src != "::/0": rule_str += " -s " + self.src if not self.multi and self.sport != "any": rule_str += " --sport " + self.sport lstr = "" if self.logtype != "": lstr = "_" + self.logtype if self.action == "allow": rule_str += " -j ACCEPT%s" % (lstr) elif self.action == "reject": rule_str += " -j REJECT%s" % (lstr) if self.protocol == "tcp": # follow TCP's default and send RST rule_str += " --reject-with tcp-reset" elif self.action == "limit": # Caller needs to change this rule_str += " -j LIMIT%s" % (lstr) else: rule_str += " -j DROP%s" % (lstr) if self.dapp != "" or self.sapp != "": # Format the comment string, and quote it just in case comment = "-m comment --comment '" pat_space = re.compile(' ') if self.dapp != "": comment += "dapp_" + pat_space.sub('%20', self.dapp) if self.dapp != "" and self.sapp != "": comment += "," if self.sapp != "": comment += "sapp_" + pat_space.sub('%20', self.sapp) comment += "'" rule_str += " " + comment return rule_str.strip() def set_action(self, action): '''Sets action of the rule''' tmp = action.lower().split('_') if tmp[0] == "allow" or tmp[0] == "reject" or tmp[0] == "limit": self.action = tmp[0] else: self.action = "deny" logtype = "" if len(tmp) > 1: logtype = tmp[1] self.set_logtype(logtype) def set_port(self, port, loc="dst"): '''Sets port and location (destination or source) of the rule''' err_msg = _("Bad port '%s'") % (port) if port == "any": pass elif loc == "dst" and self.dapp: pass elif loc == "src" and self.sapp: pass elif re.match(r'^[,:]', port) or re.match(r'[,:]$', port): raise UFWError(err_msg) elif (port.count(',') + port.count(':')) > 14: # Limitation of iptables raise UFWError(err_msg) else: ports = port.split(',') if len(ports) > 1: self.multi = True tmp = "" for p in ports: if re.match(r'^\d+:\d+$', p): # Port range self.multi = True ran = p.split(':') for q in ran: if int(q) < 1 or int(q) > 65535: raise UFWError(err_msg) if int(ran[0]) >= int(ran[1]): raise UFWError(err_msg) elif re.match('^\d+$', p): if int(p) < 1 or int(p) > 65535: raise UFWError(err_msg) elif re.match(r'^\w[\w\-]+', p): try: p = socket.getservbyname(p) except Exception: raise UFWError(err_msg) else: raise UFWError(err_msg) if tmp: tmp += "," + str(p) else: tmp = str(p) port = tmp if loc == "src": self.sport = str(port) else: self.dport = str(port) def set_protocol(self, protocol): '''Sets protocol of the rule''' if protocol in ufw.util.supported_protocols + ['any']: self.protocol = protocol else: err_msg = _("Unsupported protocol '%s'") % (protocol) raise UFWError(err_msg) def _fix_anywhere(self): '''Adjusts src and dst based on v6''' if self.v6: if self.dst and (self.dst == "any" or self.dst == "0.0.0.0/0"): self.dst = "::/0" if self.src and (self.src == "any" or self.src == "0.0.0.0/0"): self.src = "::/0" else: if self.dst and (self.dst == "any" or self.dst == "::/0"): self.dst = "0.0.0.0/0" if self.src and (self.src == "any" or self.src == "::/0"): self.src = "0.0.0.0/0" def set_v6(self, v6): '''Sets whether this is ipv6 rule, and adjusts src and dst accordingly. ''' self.v6 = v6 self._fix_anywhere() def set_src(self, addr): '''Sets source address of rule''' tmp = addr.lower() if tmp != "any" and not ufw.util.valid_address(tmp, "any"): err_msg = _("Bad source address") raise UFWError(err_msg) self.src = tmp self._fix_anywhere() def set_dst(self, addr): '''Sets destination address of rule''' tmp = addr.lower() if tmp != "any" and not ufw.util.valid_address(tmp, "any"): err_msg = _("Bad destination address") raise UFWError(err_msg) self.dst = tmp self._fix_anywhere() def set_interface(self, if_type, name): '''Sets an interface for rule''' # libxtables/xtables.c xtables_parse_interface() specifies # - < 16 # - not empty # - doesn't contain ' ' # - doesn't contain '/' # # net/core/dev.c from the kernel specifies: # - < 16 # - not empty # - != '.' or '..' # - doesn't contain '/', ':' or whitespace if if_type != "in" and if_type != "out": err_msg = _("Bad interface type") raise UFWError(err_msg) # Separate a few of the invalid checks out so we can give a nice error if '!' in str(name): err_msg = _("Bad interface name: reserved character: '!'") raise UFWError(err_msg) if ':' in str(name): err_msg = _("Bad interface name: can't use interface aliases") raise UFWError(err_msg) if str(name) == "." or str(name) == "..": err_msg = _("Bad interface name: can't use '.' or '..'") raise UFWError(err_msg) if (len(str(name)) == 0): err_msg = _("Bad interface name: interface name is empty") raise UFWError(err_msg) if (len(str(name)) > 15): err_msg = _("Bad interface name: interface name too long") raise UFWError(err_msg) # We are going to limit this even further to avoid shell meta if not re.match(r'^[a-zA-Z0-9_\-\.\+,=%@]+$', str(name)): err_msg = _("Bad interface name") raise UFWError(err_msg) if if_type == "in": self.interface_in = name else: self.interface_out = name def set_position(self, num): '''Sets the position of the rule''' # -1 prepend # 0 append # >0 insert if str(num) != "-1" and not re.match(r'^[0-9]+', str(num)): err_msg = _("Insert position '%s' is not a valid position") % (num) raise UFWError(err_msg) self.position = int(num) def set_logtype(self, logtype): '''Sets logtype of the rule''' if logtype.lower() == "log" or logtype.lower() == "log-all" or \ logtype == "": self.logtype = logtype.lower() else: err_msg = _("Invalid log type '%s'") % (logtype) raise UFWError(err_msg) def set_direction(self, direction): '''Sets direction of the rule''' if direction == "in" or direction == "out": self.direction = direction else: err_msg = _("Unsupported direction '%s'") % (direction) raise UFWError(err_msg) def get_comment(self): '''Get decoded comment of the rule''' return ufw.util.hex_decode(self.comment) def set_comment(self, comment): '''Sets comment of the rule''' self.comment = comment def normalize(self): '''Normalize src and dst to standard form''' changed = False if self.src: try: (self.src, changed) = ufw.util.normalize_address(self.src, \ self.v6) except Exception: err_msg = _("Could not normalize source address") raise UFWError(err_msg) if changed: self.updated = changed if self.dst: try: (self.dst, changed) = ufw.util.normalize_address(self.dst, \ self.v6) except Exception: err_msg = _("Could not normalize destination address") raise UFWError(err_msg) if changed: self.updated = changed if self.dport: ports = self.dport.split(',') ufw.util.human_sort(ports) self.dport = ','.join(ports) if self.sport: ports = self.sport.split(',') ufw.util.human_sort(ports) self.sport = ','.join(ports) def match(x, y): '''Check if rules match Return codes: 0 match 1 no match -1 match all but action, log-type and/or comment -2 match all but comment ''' if not x or not y: raise ValueError() dbg_msg = "No match '%s' '%s'" % (x, y) if x.dport != y.dport: debug(dbg_msg) return 1 if x.sport != y.sport: debug(dbg_msg) return 1 if x.protocol != y.protocol: debug(dbg_msg) return 1 if x.src != y.src: debug(dbg_msg) return 1 if x.dst != y.dst: debug(dbg_msg) return 1 if x.v6 != y.v6: debug(dbg_msg) return 1 if x.dapp != y.dapp: debug(dbg_msg) return 1 if x.sapp != y.sapp: debug(dbg_msg) return 1 if x.interface_in != y.interface_in: debug(dbg_msg) return 1 if x.interface_out != y.interface_out: debug(dbg_msg) return 1 if x.direction != y.direction: debug(dbg_msg) return 1 if x.forward != y.forward: debug(dbg_msg) return 1 if x.action == y.action and x.logtype == y.logtype and \ x.comment == y.comment: dbg_msg = _("Found exact match") debug(dbg_msg) return 0 if x.action == y.action and x.logtype == y.logtype and \ x.comment != y.comment: dbg_msg = _("Found exact match, excepting comment") debug(dbg_msg) return -2 dbg_msg = _("Found non-action/non-logtype/comment match " \ "(%(xa)s/%(ya)s/'%(xc)s' %(xl)s/%(yl)s/'%(yc)s')") % \ ({'xa': x.action, 'ya': y.action, 'xl': x.logtype, 'yl': y.logtype, 'xc': x.comment, 'yc': y.comment}) debug(dbg_msg) return -1 def fuzzy_dst_match(x, y): '''This will match if x is more specific than y. Eg, for protocol if x is tcp and y is all or for address if y is a network and x is a subset of y (where x is either an address or network). Returns: 0 match 1 no match -1 fuzzy match This is a fuzzy destination match, so source ports or addresses are not considered, and (currently) only incoming. ''' def _match_ports(test_p, to_match): '''Returns True if p is an exact match or within a multi rule''' if ',' in test_p or ':' in test_p: if test_p == to_match: return True return False for port in to_match.split(','): if test_p == port: return True if ':' in port: (low, high) = port.split(':') if int(test_p) >= int(low) and int(test_p) <= int(high): return True return False if not x or not y: raise ValueError() # Ok if exact match if x.match(y) == 0: return 0 dbg_msg = "No fuzzy match '%s (v6=%s)' '%s (v6=%s)'" % \ (x, x.v6, y, y.v6) # Direction must match if y.direction != "in": debug("(direction) " + dbg_msg + " (not incoming)") return 1 # forward must match if y.forward != x.forward: debug(dbg_msg + " (forward does not match)") return 1 # Protocols must match or y 'any' if x.protocol != y.protocol and y.protocol != "any": debug("(protocol) " + dbg_msg) return 1 # Destination ports must match or y 'any' if y.dport != "any" and not _match_ports(x.dport, y.dport): debug("(dport) " + dbg_msg) return 1 if y.interface_in == "": # If destination interface is not specified, destination addresses # must match or x must be contained in y if x.interface_in == "" and x._is_anywhere(x.dst): # if x and y interfaces are not specified, and x.dst is # anywhere then ok pass elif x.dst != y.dst and '/' not in y.dst: debug("(dst) " + dbg_msg) return 1 elif x.dst != y.dst and '/' in y.dst and x.v6 == y.v6 and \ not ufw.util.in_network(x.dst, y.dst, x.v6): debug("(dst) " + dbg_msg + " ('%s' not in network '%s')" % \ (x.dst, y.dst)) return 1 else: # If destination interface is specified, then: # if specified, both interfaces must match or # the IP of the interface must match the IP of y or # the IP of the interface must be contained in y if x.interface_in != "" and x.interface_in != y.interface_in: debug("(interface) " + dbg_msg + " (%s != %s)" % \ (x.interface_in, y.interface_in)) return 1 try: if_ip = ufw.util.get_ip_from_if(y.interface_in, x.v6) except IOError: debug("(interface) " + dbg_msg + " %s does not exist" % \ (y.interface_in)) return 1 if y.dst != if_ip and '/' not in y.dst: debug("(interface) " + dbg_msg + " (%s != %s)" % \ (y.dst, if_ip)) return 1 elif y.dst != if_ip and '/' in y.dst and x.v6 == y.v6 and \ not ufw.util.in_network(if_ip, y.dst, x.v6): debug("(interface) " + dbg_msg + \ " ('%s' not in network '%s')" % (if_ip, y.dst)) return 1 if x.v6 != y.v6: debug("(v6) " + dbg_msg + " (%s != %s)" % (x.dst, y.dst)) return 1 # if we made it here, it is a fuzzy match debug("(fuzzy match) '%s (v6=%s)' '%s (v6=%s)'" % (x, x.v6, y, y.v6)) return -1 def _is_anywhere(self, addr): '''Check if address is anywhere''' if addr == "::/0" or addr == "0.0.0.0/0": return True return False def get_app_tuple(self): '''Returns a tuple to identify an app rule. Tuple is: dapp dst sapp src direction_iface|direction or dport dst sapp src direction_iface|direction or dapp dst sport src direction_iface|direction where direction_iface is of form 'in_eth0', 'out_eth0' or 'in_eth0 out_eth0' (ie, both interfaces used). If no interfaces are specified, then tuple ends with the direction instead. ''' tupl = "" if self.dapp != "" or self.sapp != "": tupl = "%s %s %s %s" % (self.dapp, self.dst, self.sapp, self.src) if self.dapp == "": tupl = "%s %s %s %s" % (self.dport, self.dst, self.sapp, \ self.src) if self.sapp == "": tupl = "%s %s %s %s" % (self.dapp, self.dst, self.sport, \ self.src) # if neither interface exists, add the direction if self.interface_in == "" and self.interface_out == "": tupl += " %s" % (self.direction) # otherwise, add the interfaces else: if self.interface_in != "": tupl += " in_%s" % (self.interface_in) if self.interface_out != "": tupl += " out_%s" % (self.interface_out) return tupl def verify(self, rule_iptype): '''Verify rule''' # Verify protocol not specified with application rule if self.protocol != "any" and \ (self.sapp != "" or self.dapp != ""): err_msg = _("Improper rule syntax ('%s' specified with app rule)") \ % (self.protocol) raise UFWError(err_msg) if self.protocol in ufw.util.ipv4_only_protocols and \ rule_iptype == "v6": # Can't use protocol these protocols with v6 addresses err_msg = _("Invalid IPv6 address with protocol '%s'") % \ (self.protocol) raise UFWError(err_msg) if self.protocol in ufw.util.portless_protocols: if self.dport != "any" or self.sport != "any": err_msg = _("Invalid port with protocol '%s'") % \ (self.protocol) raise UFWError(err_msg)