# # The Qubes OS Project, http://www.qubes-os.org # # Copyright (C) 2011 Tomasz Sterna # # 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 Lesser General Public License along # with this program; if not, see . # # import datetime import re from PyQt5 import QtCore, QtGui, QtWidgets # pylint: disable=import-error import qubesadmin.firewall from . import ui_newfwruledlg # pylint: disable=no-name-in-module class FirewallModifiedOutsideError(ValueError): pass class NewFwRuleDlg(QtWidgets.QDialog, ui_newfwruledlg.Ui_NewFwRuleDlg): def __init__(self, parent=None): super(NewFwRuleDlg, self).__init__(parent) self.setupUi(self) self.set_ok_state(False) self.addressComboBox.editTextChanged.connect( self.address_editing_finished) self.serviceComboBox.setValidator(QtGui.QRegExpValidator( QtCore.QRegExp("[a-z][a-z0-9-]+|[0-9]+(-[0-9]+)?", QtCore.Qt.CaseInsensitive), None)) self.serviceComboBox.setEnabled(False) self.serviceComboBox.setInsertPolicy(QtWidgets.QComboBox.InsertAtBottom) self.populate_combos() self.serviceComboBox.setInsertPolicy(QtWidgets.QComboBox.InsertAtTop) self.model = None def try_to_create_rule(self): # return True if successful, False otherwise address = str(self.addressComboBox.currentText().strip()) service = str(self.serviceComboBox.currentText().strip()) rule = qubesadmin.firewall.Rule(None, action='accept') if address is not None and address != "*": try: rule.dsthost = address except ValueError: QtWidgets.QMessageBox.warning( self, self.tr("Invalid address"), self.tr("Address '{0}' is invalid.").format(address)) return False if self.tcp_radio.isChecked(): rule.proto = 'tcp' elif self.udp_radio.isChecked(): rule.proto = 'udp' if self.model.port_range_pattern.fullmatch(service): try: rule.dstports = service except ValueError: QtWidgets.QMessageBox.warning( self, self.tr("Invalid port or service"), self.tr("Port number or service '{0}' is " "invalid.").format(service)) return False elif service: if self.model.service_port_pattern.fullmatch(service): parsed_service = self.model.service_port_pattern.match( service).groups()[0] else: parsed_service = service try: rule.dstports = parsed_service except (TypeError, ValueError): if self.model.get_service_port(parsed_service) is not None: rule.dstports = self.model.get_service_port(parsed_service) else: QtWidgets.QMessageBox.warning( self, self.tr("Invalid port or service"), self.tr( "Port number or service '{0}' is " "invalid.".format(parsed_service))) return False if self.model.current_row is not None: self.model.set_child(self.model.current_row, rule) else: self.model.append_child(rule) return True def accept(self): if self.tcp_radio.isChecked() or self.udp_radio.isChecked(): if not self.serviceComboBox.currentText(): msg = QtWidgets.QMessageBox() msg.warning( self, self.tr("Firewall rule"), self.tr("You need to fill service " "name/port for TCP/UDP rule")) return if self.try_to_create_rule(): super().accept() def populate_combos(self): example_addresses = [ "", "www.example.com", "192.168.1.100", "192.168.0.0/16", "*" ] example_services = [ '', '22', '80', '1024-1234', 'http', 'https', 'ftp', 'ftps', 'smtp', 'pop3', 'pop3s', 'imap', 'imaps', 'odmr', 'nntp', 'nntps', 'ssh', 'telnet', 'telnets', 'ntp', 'snmp', 'ldap', 'ldaps', 'irc', 'ircs-u', 'xmpp-client', 'syslog', 'printer', 'nfs', 'x11' ] for address in example_addresses: self.addressComboBox.addItem(address) for service in example_services: self.serviceComboBox.addItem(service) def address_editing_finished(self): self.set_ok_state(True) def set_ok_state(self, ok_state): ok_button = self.buttonBox.button(QtWidgets.QDialogButtonBox.Ok) if ok_button is not None: ok_button.setEnabled(ok_state) def on_tcp_radio_toggled(self, checked): if checked: self.serviceComboBox.setEnabled(True) def on_udp_radio_toggled(self, checked): if checked: self.serviceComboBox.setEnabled(True) def on_any_radio_toggled(self, checked): if checked: self.serviceComboBox.setEnabled(False) class QubesFirewallRulesModel(QtCore.QAbstractItemModel): def __init__(self, parent=None): QtCore.QAbstractItemModel.__init__(self, parent) self.current_row = None self.__column_names = {0: "Address", 1: "Port/Service", 2: "Protocol", } self.__services = list() self.port_range_pattern = re.compile(r'\d+-\d+') self.service_port_pattern = re.compile(r'(\d*) \([a-zA-Z0-9-]*\)') pattern = re.compile( r"(?P[a-z][a-z0-9-]+)\s+(?P[0-9]+)/" r"(?P[a-z]+)", re.IGNORECASE) with open('/etc/services', 'r', encoding='utf-8') as file: for line in file: match = pattern.match(line) if match is not None: service = match.groupdict() self.__services.append( (service["name"], int(service["port"]),)) self.fw_changed = False self.allow = None # is the default policy allow or deny self.temp_full_access_expire_time = None # temporary full access time self.__vm = None # VM that the model applies to self.__children = None # list of rules in the FW def sort(self, idx, order): rev = (order == QtCore.Qt.AscendingOrder) self.children.sort(key=lambda x: self.get_column_string(idx, x), reverse=rev) index1 = self.createIndex(0, 0) index2 = self.createIndex(len(self) - 1, len(self.__column_names) - 1) self.dataChanged.emit(index1, index2) def get_service_name(self, port): for service in self.__services: if str(service[1]) == str(port): return "{0} ({1})".format(str(port), service[0]) return str(port) def get_service_port(self, name): for service in self.__services: if service[0] == name: return service[1] return None def get_column_string(self, col, rule): # pylint: disable=too-many-return-statements # Address if col == 0: if rule.dsthost is None: return "*" if rule.dsthost.type == 'dst4' and rule.dsthost.prefixlen == '32': return str(rule.dsthost)[:-3] if rule.dsthost.type == 'dst6' and rule.dsthost.prefixlen == '128': return str(rule.dsthost)[:-4] return str(rule.dsthost) # Service if col == 1: if rule.dstports is None: return "any" if rule.dstports.range[0] != rule.dstports.range[1]: return str(rule.dstports) return self.get_service_name(rule.dstports) # Protocol if col == 2: if rule.proto is None: return "any" return str(rule.proto) return "unknown" def get_firewall_conf(self, vm): conf = { 'allow': None, 'expire': 0, 'rules': [], } allow_dns = False allow_icmp = False common_action = None reversed_rules = reversed(vm.firewall.rules) last_rule = next(reversed_rules, None) if last_rule is None: raise FirewallModifiedOutsideError( self.tr('At least one rule must exist.')) if last_rule == qubesadmin.firewall.Rule('action=accept') \ or last_rule == qubesadmin.firewall.Rule('action=drop'): common_action = last_rule.action else: raise FirewallModifiedOutsideError( self.tr('Last rule must be either drop all or accept all.')) dns_rule = qubesadmin.firewall.Rule(None, action='accept', specialtarget='dns') icmp_rule = qubesadmin.firewall.Rule(None, action='accept', proto='icmp') for rule in reversed_rules: if rule == dns_rule: allow_dns = True continue if rule == icmp_rule: allow_icmp = True continue if rule.specialtarget is not None or rule.icmptype is not None: raise FirewallModifiedOutsideError( self.tr("Rule type unknown!")) if (rule.dsthost is not None or rule.proto is not None) \ and rule.expire is None: if rule.action == 'accept': conf['rules'].insert(0, rule) continue raise FirewallModifiedOutsideError( self.tr('No blacklist support.')) if rule.expire is not None and rule.dsthost is None \ and rule.proto is None: conf['expire'] = int(str(rule.expire)) continue raise FirewallModifiedOutsideError(self.tr('it does not add up.')) conf['allow'] = (common_action == 'accept') if not allow_icmp and not conf['allow']: raise FirewallModifiedOutsideError(self.tr('ICMP must be allowed.')) if not allow_dns and not conf['allow']: raise FirewallModifiedOutsideError(self.tr('DNS must be allowed')) return conf @staticmethod def write_firewall_conf(vm, conf): rules = [] for rule in conf['rules']: rules.append(rule) if not conf['allow']: rules.append(qubesadmin.firewall.Rule( None, action='accept', specialtarget='dns')) if not conf['allow']: rules.append(qubesadmin.firewall.Rule( None, action='accept', proto='icmp')) if conf['allow']: rules.append(qubesadmin.firewall.Rule( None, action='accept')) else: rules.append(qubesadmin.firewall.Rule( None, action='drop')) vm.firewall.rules = rules def set_vm(self, vm): self.__vm = vm self.clear_children() conf = self.get_firewall_conf(vm) self.allow = conf["allow"] self.temp_full_access_expire_time = conf['expire'] for rule in conf["rules"]: self.append_child(rule) def get_vm_name(self): return self.__vm.name def apply_rules(self, allow, temp_full_access=False, temp_full_access_time=None): assert self.__vm is not None if self.allow != allow or \ (self.temp_full_access_expire_time != 0) != temp_full_access: self.fw_changed = True conf = {"allow": allow, "rules": list() } conf['rules'].extend(self.children) if temp_full_access and not allow: conf["rules"].append(qubesadmin.firewall.Rule( None, action='accept', expire=int(datetime.datetime.now().strftime("%s")) + temp_full_access_time * 60)) if self.fw_changed: self.write_firewall_conf(self.__vm, conf) def populate_edit_dialog(self, dialog, row): address = self.get_column_string(0, self.children[row]) dialog.addressComboBox.setItemText(0, address) dialog.addressComboBox.setCurrentIndex(0) service = self.get_column_string(1, self.children[row]) if service == "any": service = "" dialog.serviceComboBox.setItemText(0, service) dialog.serviceComboBox.setCurrentIndex(0) protocol = self.get_column_string(2, self.children[row]) if protocol == "tcp": dialog.tcp_radio.setChecked(True) elif protocol == "udp": dialog.udp_radio.setChecked(True) else: dialog.any_radio.setChecked(True) def run_rule_dialog(self, dialog, row=None): self.current_row = row dialog.model = self dialog.exec() def index(self, row, column, parent=QtCore.QModelIndex()): if not self.hasIndex(row, column, parent): return QtCore.QModelIndex() return self.createIndex(row, column, self.children[row]) def parent(self, child): # pylint: disable=unused-argument,no-self-use return QtCore.QModelIndex() # pylint: disable=invalid-name,unused-argument def rowCount(self, parent=QtCore.QModelIndex()): return len(self) # pylint: disable=invalid-name,unused-argument def columnCount(self, parent=QtCore.QModelIndex()): return len(self.__column_names) # pylint: disable=invalid-name,no-self-use def hasChildren(self, index=QtCore.QModelIndex()): parent_item = index.internalPointer() return parent_item is None def data(self, index, role=QtCore.Qt.DisplayRole): if index.isValid() and role == QtCore.Qt.DisplayRole: return self.get_column_string(index.column(), self.children[index.row()]) # pylint: disable=invalid-name def headerData(self, section, orientation, role=QtCore.Qt.DisplayRole): if section < len(self.__column_names) \ and orientation == QtCore.Qt.Horizontal \ and role == QtCore.Qt.DisplayRole: return self.__column_names[section] @property def children(self): return self.__children def append_child(self, child): row = len(self) self.beginInsertRows(QtCore.QModelIndex(), row, row) self.children.append(child) self.endInsertRows() index = self.createIndex(row, 0, child) self.dataChanged.emit(index, index) self.fw_changed = True def remove_child(self, i): if i >= len(self): return self.beginRemoveRows(QtCore.QModelIndex(), i, i) del self.children[i] self.endRemoveRows() index = self.createIndex(i, 0) self.dataChanged.emit(index, index) self.fw_changed = True def set_child(self, i, child): self.children[i] = child index = self.createIndex(i, 0, child) self.dataChanged.emit(index, index) self.fw_changed = True def clear_children(self): self.__children = list() def __len__(self): return len(self.children)