Gecko [ sachasauda.akalacademy.ac.in ]

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

Code Editor : 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

${this.title}

Code Editor : fw_ipset.py