Source code for qubesadmin.firewall

# -*- encoding: utf8 -*-
# pylint: disable=too-few-public-methods
#
# The Qubes OS Project, http://www.qubes-os.org
#
# Copyright (C) 2017 Marek Marczykowski-Górecki
#                               <marmarek@invisiblethingslab.com>
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU Lesser General Public License as published by
# the Free Software Foundation; either version 2.1 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 Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public License along
# with this program; if not, see <http://www.gnu.org/licenses/>.

'''Firewall configuration interface'''

import datetime
import socket
import string


[docs]class RuleOption(object): '''Base class for a single rule element''' def __init__(self, value): self._value = str(value) @property def rule(self): '''API representation of this rule element''' raise NotImplementedError @property def pretty_value(self): '''Human readable representation''' return str(self) def __str__(self): return self._value def __eq__(self, other): return str(self) == other
# noinspection PyAbstractClass
[docs]class RuleChoice(RuleOption): '''Base class for multiple-choices rule elements''' # pylint: disable=abstract-method def __init__(self, value): super(RuleChoice, self).__init__(value) self.allowed_values = \ [v for k, v in self.__class__.__dict__.items() if not k.startswith('__') and isinstance(v, str) and not v.startswith('__')] if value not in self.allowed_values: raise ValueError(value)
[docs]class Action(RuleChoice): '''Rule action''' accept = 'accept' drop = 'drop' @property def rule(self): '''API representation of this rule element''' return 'action=' + str(self)
[docs]class Proto(RuleChoice): '''Protocol name''' tcp = 'tcp' udp = 'udp' icmp = 'icmp' @property def rule(self): '''API representation of this rule element''' return 'proto=' + str(self)
[docs]class DstHost(RuleOption): '''Represent host/network address: either IPv4, IPv6, or DNS name''' def __init__(self, value, prefixlen=None): # TODO: in python >= 3.3 ipaddress module could be used if value.count('/') > 1: raise ValueError('Too many /: ' + value) if not value.count('/'): # add prefix length to bare IP addresses try: socket.inet_pton(socket.AF_INET6, value) if prefixlen is not None: self.prefixlen = prefixlen else: self.prefixlen = 128 if self.prefixlen < 0 or self.prefixlen > 128: raise ValueError( 'netmask for IPv6 must be between 0 and 128') value += '/' + str(self.prefixlen) self.type = 'dst6' except socket.error: try: socket.inet_pton(socket.AF_INET, value) if value.count('.') != 3: raise ValueError( 'Invalid number of dots in IPv4 address') if prefixlen is not None: self.prefixlen = prefixlen else: self.prefixlen = 32 if self.prefixlen < 0 or self.prefixlen > 32: raise ValueError( 'netmask for IPv4 must be between 0 and 32') value += '/' + str(self.prefixlen) self.type = 'dst4' except socket.error: self.type = 'dsthost' self.prefixlen = 0 safe_set = string.ascii_lowercase + string.digits + '-._' if not all(c in safe_set for c in value): raise ValueError('Invalid hostname') else: host, prefixlen = value.split('/', 1) prefixlen = int(prefixlen) if prefixlen < 0: raise ValueError('netmask must be non-negative') self.prefixlen = prefixlen try: socket.inet_pton(socket.AF_INET6, host) if prefixlen > 128: raise ValueError('netmask for IPv6 must be <= 128') self.type = 'dst6' except socket.error: try: socket.inet_pton(socket.AF_INET, host) if prefixlen > 32: raise ValueError('netmask for IPv4 must be <= 32') self.type = 'dst4' if host.count('.') != 3: raise ValueError( 'Invalid number of dots in IPv4 address') except socket.error: raise ValueError('Invalid IP address: ' + host) super(DstHost, self).__init__(value) @property def rule(self): '''API representation of this rule element''' if self.prefixlen == 0 and self.type != 'dsthost': # 0.0.0.0/0 or ::/0, doesn't limit to any particular host, # so skip it return None return self.type + '=' + str(self)
[docs]class DstPorts(RuleOption): '''Destination port(s), for TCP/UDP only''' def __init__(self, value): if isinstance(value, int): value = str(value) if value.count('-') == 1: self.range = [int(x) for x in value.split('-', 1)] elif not value.count('-'): self.range = [int(value), int(value)] else: raise ValueError(value) if any(port < 0 or port > 65536 for port in self.range): raise ValueError('Ports out of range') if self.range[0] > self.range[1]: raise ValueError('Invalid port range') super(DstPorts, self).__init__( str(self.range[0]) if self.range[0] == self.range[1] else '{!s}-{!s}'.format(*self.range)) @property def rule(self): '''API representation of this rule element''' return 'dstports=' + '{!s}-{!s}'.format(*self.range)
[docs]class IcmpType(RuleOption): '''ICMP packet type''' def __init__(self, value): super(IcmpType, self).__init__(value) value = int(value) if value < 0 or value > 255: raise ValueError('ICMP type out of range') @property def rule(self): '''API representation of this rule element''' return 'icmptype=' + str(self)
[docs]class SpecialTarget(RuleChoice): '''Special destination''' dns = 'dns' @property def rule(self): '''API representation of this rule element''' return 'specialtarget=' + str(self)
[docs]class Expire(RuleOption): '''Rule expire time''' def __init__(self, value): super(Expire, self).__init__(value) self.datetime = datetime.datetime.utcfromtimestamp(int(value)) @property def rule(self): '''API representation of this rule element''' return 'expire=' + str(self) @property def expired(self): '''Has this rule expired already?''' return self.datetime < datetime.datetime.utcnow() @property def pretty_value(self): '''Human readable representation''' now = datetime.datetime.utcnow() duration = (self.datetime - now).total_seconds() return "{:+.0f}s".format(duration)
[docs]class Comment(RuleOption): '''User comment''' @property def rule(self): '''API representation of this rule element''' return 'comment=' + str(self)
[docs]class Rule(object): '''A single firewall rule''' def __init__(self, rule, **kwargs): '''Single firewall rule :param xml: XML element describing rule, or None :param kwargs: rule elements ''' self._action = None self._proto = None self._dsthost = None self._dstports = None self._icmptype = None self._specialtarget = None self._expire = None self._comment = None rule_dict = {} if rule is not None: rule_opts, _, comment = rule.partition('comment=') rule_dict = dict(rule_opt.split('=', 1) for rule_opt in rule_opts.split(' ') if rule_opt) if comment: rule_dict['comment'] = comment rule_dict.update(kwargs) rule_elements = ('action', 'proto', 'dsthost', 'dst4', 'dst6', 'specialtarget', 'dstports', 'icmptype', 'expire', 'comment') for rule_opt in rule_elements: value = rule_dict.pop(rule_opt, None) if value is None: continue if rule_opt in ('dst4', 'dst6'): rule_opt = 'dsthost' setattr(self, rule_opt, value) if rule_dict: raise ValueError('Unknown rule elements: {!r}'.format( rule_dict)) if self.action is None: raise ValueError('missing action=') @property def action(self): '''rule action''' return self._action @action.setter def action(self, value): if not isinstance(value, Action): value = Action(value) self._action = value @property def proto(self): '''protocol to match''' return self._proto @proto.setter def proto(self, value): if value is not None and not isinstance(value, Proto): value = Proto(value) if value not in ('tcp', 'udp'): self.dstports = None if value not in ('icmp',): self.icmptype = None self._proto = value @property def dsthost(self): '''destination host/network''' return self._dsthost @dsthost.setter def dsthost(self, value): if value is not None and not isinstance(value, DstHost): value = DstHost(value) self._dsthost = value @property def dstports(self): ''''Destination port(s) (for \'tcp\' and \'udp\' protocol only)''' return self._dstports @dstports.setter def dstports(self, value): if value is not None: if self.proto not in ('tcp', 'udp'): raise ValueError( 'dstports valid only for \'tcp\' and \'udp\' protocols') if not isinstance(value, DstPorts): value = DstPorts(value) self._dstports = value @property def icmptype(self): '''ICMP packet type (for \'icmp\' protocol only)''' return self._icmptype @icmptype.setter def icmptype(self, value): if value is not None: if self.proto not in ('icmp',): raise ValueError('icmptype valid only for \'icmp\' protocol') if not isinstance(value, IcmpType): value = IcmpType(value) self._icmptype = value @property def specialtarget(self): '''Special target, for now only \'dns\' supported''' return self._specialtarget @specialtarget.setter def specialtarget(self, value): if not isinstance(value, SpecialTarget): value = SpecialTarget(value) self._specialtarget = value @property def expire(self): '''Timestamp (UNIX epoch) on which this rule expire''' return self._expire @expire.setter def expire(self, value): if not isinstance(value, Expire): value = Expire(value) self._expire = value @property def comment(self): '''User comment''' return self._comment @comment.setter def comment(self, value): if not isinstance(value, Comment): value = Comment(value) self._comment = value @property def rule(self): '''API representation of this rule''' values = [] # comment must be the last one for prop in ('action', 'proto', 'dsthost', 'dstports', 'icmptype', 'specialtarget', 'expire', 'comment'): value = getattr(self, prop) if value is None: continue if value.rule is None: continue values.append(value.rule) return ' '.join(values) def __eq__(self, other): if isinstance(other, Rule): return self.rule == other.rule if isinstance(other, str): return self.rule == str return NotImplemented def __repr__(self): return 'Rule(\'{}\')'.format(self.rule)
[docs]class Firewall(object): '''Firewal manager for a VM''' def __init__(self, vm): self.vm = vm self._rules = [] self._policy = None self._loaded = False
[docs] def load_rules(self): '''Force (re-)loading firewall rules''' rules_str = self.vm.qubesd_call(None, 'admin.vm.firewall.Get') rules = [] for rule_str in rules_str.decode().splitlines(): rules.append(Rule(rule_str)) self._rules = rules self._loaded = True
@property def rules(self): '''Firewall rules You can either copy them, edit and then assign new rules list to this property, or edit in-place and call :py:meth:`save_rules`. Once rules are loaded, they are cached. To reload rules, call :py:meth:`load_rules`. ''' if not self._loaded: self.load_rules() return self._rules @rules.setter def rules(self, value): self.save_rules(value) self._rules = value
[docs] def save_rules(self, rules=None): '''Save firewall rules. Needs to be called after in-place editing :py:attr:`rules`. ''' if rules is None: rules = self._rules self.vm.qubesd_call(None, 'admin.vm.firewall.Set', payload=(''.join('{}\n'.format(rule.rule) for rule in rules)).encode('ascii'))
@property def policy(self): '''Default action to take if no rule matches''' return Action('drop')
[docs] def reload(self): '''Force reload the same firewall rules. Can be used for example to force again names resolution. ''' self.vm.qubesd_call(None, 'admin.vm.firewall.Reload')