golden hour

/lib/python2.7/site-packages/firewall/core

File/Folder	Size	Actions
__init__.py	0 B
__init__.pyc	145 B
__init__.pyo	145 B
base.py	1.94 KB
base.pyc	1.29 KB
base.pyo	1.29 KB
ebtables.py	9.13 KB
ebtables.pyc	9.04 KB
ebtables.pyo	9.04 KB
fw.py	43.71 KB
fw.pyc	30.67 KB
fw.pyo	30.67 KB
fw_config.py	35.99 KB
fw_config.pyc	30.69 KB
fw_config.pyo	30.69 KB
fw_direct.py	20.12 KB
fw_direct.pyc	14.77 KB
fw_direct.pyo	14.77 KB
fw_helper.py	1.79 KB
fw_helper.pyc	2.57 KB
fw_helper.pyo	2.57 KB
fw_icmptype.py	2.77 KB
fw_icmptype.pyc	3 KB
fw_icmptype.pyo	3 KB
fw_ifcfg.py	2.5 KB
fw_ifcfg.pyc	1.84 KB
fw_ifcfg.pyo	1.84 KB
fw_ipset.py	8.96 KB
fw_ipset.pyc	9.02 KB
fw_ipset.pyo	9.02 KB
fw_nm.py	6.49 KB
fw_nm.pyc	5.93 KB
fw_nm.pyo	5.93 KB
fw_policies.py	2.74 KB
fw_policies.pyc	2.94 KB
fw_policies.pyo	2.94 KB
fw_service.py	1.6 KB
fw_service.pyc	2.14 KB
fw_service.pyo	2.14 KB
fw_test.py	22.06 KB
fw_test.pyc	17.45 KB
fw_test.pyo	17.45 KB
fw_transaction.py	10.54 KB
fw_transaction.pyc	10.96 KB
fw_transaction.pyo	10.96 KB
fw_zone.py	75.6 KB
fw_zone.pyc	57.31 KB
fw_zone.pyo	57.31 KB
helper.py	804 B
helper.pyc	222 B
helper.pyo	222 B
icmp.py	3.03 KB
icmp.pyc	2.89 KB
icmp.pyo	2.89 KB
io	-
ipXtables.py	47.68 KB
ipXtables.pyc	34.8 KB
ipXtables.pyo	34.8 KB
ipset.py	9.1 KB
ipset.pyc	9.15 KB
ipset.pyo	9.15 KB
logger.py	30.31 KB
logger.pyc	27.43 KB
logger.pyo	27.43 KB
modules.py	3.63 KB
modules.pyc	3.56 KB
modules.pyo	3.56 KB
nftables.py	60.55 KB
nftables.pyc	38.56 KB
nftables.pyo	38.56 KB
prog.py	1.47 KB
prog.pyc	988 B
prog.pyo	988 B
rich.py	29.34 KB
rich.pyc	23.73 KB
rich.pyo	23.73 KB
watcher.py	3.15 KB
watcher.pyc	3.55 KB
watcher.pyo	3.55 KB

Edit: fw_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 backend"""

__all__ = [ "FirewallIPSet" ]

from firewall.core.logger import log
from firewall.core.ipset import remove_default_create_options as rm_def_cr_opts
from firewall.core.io.ipset import IPSet
from firewall import errors
from firewall.errors import FirewallError

class FirewallIPSet(object):
    def __init__(self, fw):
        self._fw = fw
        self._ipsets = { }

def __repr__(self):
        return '%s(%r)' % (self.__class__, self._ipsets)

# ipsets

def cleanup(self):
        self._ipsets.clear()

def check_ipset(self, name):
        if name not in self.get_ipsets():
            raise FirewallError(errors.INVALID_IPSET, name)

def query_ipset(self, name):
        return name in self.get_ipsets()

def get_ipsets(self):
        return sorted(self._ipsets.keys())

def has_ipsets(self):
        return len(self._ipsets) > 0

def get_ipset(self, name, applied=False):
        self.check_ipset(name)
        obj = self._ipsets[name]
        if applied:
            self.check_applied_obj(obj)
        return obj

def _error2warning(self, f, name, *args):
        # transform errors into warnings
        try:
            f(name, *args)
        except FirewallError as error:
            msg = str(error)
            log.warning("%s: %s" % (name, msg))

def backends(self):
        backends = []
        if self._fw.nftables_enabled:
            backends.append(self._fw.nftables_backend)
        if self._fw.ipset_enabled:
            backends.append(self._fw.ipset_backend)
        return backends

def add_ipset(self, obj):
        if obj.type not in self._fw.ipset_supported_types:
            raise FirewallError(errors.INVALID_TYPE,
                                "'%s' is not supported by ipset." % obj.type)
        self._ipsets[obj.name] = obj

def remove_ipset(self, name, keep=False):
        obj = self._ipsets[name]
        if obj.applied and not keep:
            try:
                for backend in self.backends():
                    backend.set_destroy(name)
            except Exception as msg:
                raise FirewallError(errors.COMMAND_FAILED, msg)
        else:
            log.debug1("Keeping ipset '%s' because of timeout option", name)
        del self._ipsets[name]

def apply_ipsets(self):
        for name in self.get_ipsets():
            obj = self._ipsets[name]
            obj.applied = False

log.debug1("Applying ipset '%s'" % name)

for backend in self.backends():
                if backend.name == "ipset":
                    active = backend.set_get_active_terse()

if name in active and ("timeout" not in obj.options or \
                                           obj.options["timeout"] == "0" or \
                                           obj.type != active[name][0] or \
                                           rm_def_cr_opts(obj.options) != \
                                           active[name][1]):
                        try:
                            backend.set_destroy(name)
                        except Exception as msg:
                            raise FirewallError(errors.COMMAND_FAILED, msg)

if self._fw.individual_calls() \
                   or backend.name == "nftables":
                    try:
                        backend.set_create(obj.name, obj.type, obj.options)
                    except Exception as msg:
                        raise FirewallError(errors.COMMAND_FAILED, msg)
                    else:
                        obj.applied = True
                        if "timeout" in obj.options and \
                           obj.options["timeout"] != "0":
                            # no entries visible for ipsets with timeout
                            continue

for entry in obj.entries:
                        try:
                            backend.set_add(obj.name, entry)
                        except Exception as msg:
                            raise FirewallError(errors.COMMAND_FAILED, msg)
                else:
                    try:
                        backend.set_restore(obj.name, obj.type,
                                                       obj.entries, obj.options,
                                                       None)
                    except Exception as msg:
                        raise FirewallError(errors.COMMAND_FAILED, msg)
                    else:
                        obj.applied = True

# TYPE

def get_type(self, name):
        return self.get_ipset(name, applied=True).type

# DIMENSION
    def get_dimension(self, name):
        return len(self.get_ipset(name, applied=True).type.split(","))

def check_applied(self, name):
        obj = self.get_ipset(name)
        self.check_applied_obj(obj)

def check_applied_obj(self, obj):
        if not obj.applied:
            raise FirewallError(
                errors.NOT_APPLIED, obj.name)

# OPTIONS

def get_family(self, name):
        obj = self.get_ipset(name, applied=True)
        if "family" in obj.options:
            if obj.options["family"] == "inet6":
                return "ipv6"
        return "ipv4"

# ENTRIES

def __entry_id(self, entry):
        return entry

def __entry(self, enable, name, entry):
        pass

def add_entry(self, name, entry):
        obj = self.get_ipset(name, applied=True)

IPSet.check_entry(entry, obj.options, obj.type)
        if entry in obj.entries:
            raise FirewallError(errors.ALREADY_ENABLED,
                                "'%s' already is in '%s'" % (entry, name))

try:
            for backend in self.backends():
                backend.set_add(obj.name, entry)
        except Exception as msg:
            raise FirewallError(errors.COMMAND_FAILED, msg)
        else:
            if "timeout" not in obj.options or obj.options["timeout"] == "0" \
               and entry not in obj.entries:
                # no entries visible for ipsets with timeout
                obj.entries.append(entry)

def remove_entry(self, name, entry):
        obj = self.get_ipset(name, applied=True)

# no entry check for removal
        if entry not in obj.entries:
            raise FirewallError(errors.NOT_ENABLED,
                                "'%s' not in '%s'" % (entry, name))
        try:
            for backend in self.backends():
                backend.set_delete(obj.name, entry)
        except Exception as msg:
            raise FirewallError(errors.COMMAND_FAILED, msg)
        else:
            if "timeout" not in obj.options or obj.options["timeout"] == "0" \
               and entry not in obj.entries:
                # no entries visible for ipsets with timeout
                obj.entries.remove(entry)

def query_entry(self, name, entry):
        obj = self.get_ipset(name, applied=True)
        if "timeout" in obj.options and obj.options["timeout"] != "0":
            # no entries visible for ipsets with timeout
            raise FirewallError(errors.IPSET_WITH_TIMEOUT, name)

return entry in obj.entries

def get_entries(self, name):
        obj = self.get_ipset(name, applied=True)
        return obj.entries

def set_entries(self, name, entries):
        obj = self.get_ipset(name, applied=True)

for entry in entries:
            IPSet.check_entry(entry, obj.options, obj.type)
        if "timeout" not in obj.options or obj.options["timeout"] == "0":
            # no entries visible for ipsets with timeout
            obj.entries = entries

try:
            for backend in self.backends():
                backend.set_flush(obj.name)
        except Exception as msg:
            raise FirewallError(errors.COMMAND_FAILED, msg)
        else:
            obj.applied = True

try:
            for backend in self.backends():
                if self._fw.individual_calls() \
                   or backend.name == "nftables":
                    for entry in obj.entries:
                        backend.set_add(obj.name, entry)
                else:
                    backend.set_restore(obj.name, obj.type, obj.entries,
                                                   obj.options, None)
        except Exception as msg:
            raise FirewallError(errors.COMMAND_FAILED, msg)
        else:
            obj.applied = True

return