# of using script templates.
from os import path
+from formencode import Invalid
+from formencode.validators import OneOf, CIDR, Int
+import logging ; log = logging.getLogger('pymin.services.firewall')
-from pymin.seqtools import Sequence
+from pymin.item import Item
+from pymin.validatedclass import Field
from pymin.dispatcher import Handler, handler, HandlerError
from pymin.services.util import Restorable, ConfigWriter, ServiceHandler, \
- TransactionalHandler, SubHandler
+ TransactionalHandler, ListSubHandler
-__ALL__ = ('FirewallHandler', 'Error', 'RuleError', 'RuleAlreadyExistsError',
- 'RuleNotFoundError')
+__all__ = ('FirewallHandler',)
-class Error(HandlerError):
- r"""
- Error(command) -> Error instance :: Base FirewallHandler exception class.
+class UpOneOf(OneOf):
+ def validate_python(self, value, state):
+ value = value.upper()
+ return OneOf.validate_python(self, value, state)
- All exceptions raised by the FirewallHandler inherits from this one, so you can
- easily catch any FirewallHandler exception.
-
- message - A descriptive error message.
- """
- pass
-
-class RuleError(Error, KeyError):
- r"""
- RuleError(rule) -> RuleError instance.
-
- This is the base exception for all rule related errors.
- """
-
- def __init__(self, rule):
- r"Initialize the object. See class documentation for more info."
- self.message = u'Rule error: "%s"' % rule
-
-class RuleAlreadyExistsError(RuleError):
- r"""
- RuleAlreadyExistsError(rule) -> RuleAlreadyExistsError instance.
-
- This exception is raised when trying to add a rule that already exists.
- """
-
- def __init__(self, rule):
- r"Initialize the object. See class documentation for more info."
- self.message = u'Rule already exists: "%s"' % rule
-
-class RuleNotFoundError(RuleError):
- r"""
- RuleNotFoundError(rule) -> RuleNotFoundError instance.
-
- This exception is raised when trying to operate on a rule that doesn't
- exists.
- """
-
- def __init__(self, rule):
- r"Initialize the object. See class documentation for more info."
- self.message = u'Rule not found: "%s"' % rule
-
-
-class Rule(Sequence):
+class Rule(Item):
r"""Rule(chain, target[, src[, dst[, ...]]]) -> Rule instance.
chain - INPUT, OUTPUT or FORWARD.
src_port - Source port (only for UDP or TCP protocols).
dst_port - Destination port (only for UDP or TCP protocols).
"""
-
- def __init__(self, chain, target, src=None, dst=None, protocol=None,
- src_port=None, dst_port=None):
- r"Initialize object, see class documentation for details."
- self.chain = chain
- self.target = target
- self.src = src
- self.dst = dst
- self.protocol = protocol
- # TODO Validate that src_port and dst_port could be not None only
- # if the protocol is UDP or TCP
- self.src_port = src_port
- self.dst_port = dst_port
-
- def update(self, chain=None, target=None, src=None, dst=None, protocol=None,
- src_port=None, dst_port=None):
- r"update([chain[, ...]]) -> Update the values of a rule (see Rule doc)."
- if chain is not None: self.chain = chain
- if target is not None: self.target = target
- if src is not None: self.src = src
- if dst is not None: self.dst = dst
- if protocol is not None: self.protocol = protocol
- # TODO Validate that src_port and dst_port could be not None only
- # if the protocol is UDP or TCP
- if src_port is not None: self.src_port = src_port
- if dst_port is not None: self.dst_port = dst_port
-
- def __cmp__(self, other):
- r"Compares two Rule objects."
- if self.chain == other.chain \
- and self.target == other.target \
- and self.src == other.src \
- and self.dst == other.dst \
- and self.protocol == other.protocol \
- and self.src_port == other.src_port \
- and self.dst_port == other.dst_port:
- return 0
- return cmp(id(self), id(other))
-
- def as_tuple(self):
- r"Return a tuple representing the rule."
- return (self.chain, self.target, self.src, self.dst, self.protocol,
- self.src_port, self.dst_port)
-
-class RuleHandler(SubHandler):
- r"""RuleHandler(rules) -> RuleHandler instance :: Handle a list of rules.
+ chain = Field(UpOneOf(['INPUT', 'OUTPUT', 'FORWARD'], not_empty=True))
+ target = Field(UpOneOf(['ACCEPT', 'REJECT', 'DROP'], not_empty=True))
+ src = Field(CIDR(if_empty=None, if_missing=None))
+ dst = Field(CIDR(if_empty=None, if_missing=None))
+ protocol = Field(UpOneOf(['ICMP', 'UDP', 'TCP', 'ALL'], if_missing=None))
+ src_port = Field(Int(min=0, max=65535, if_empty=None, if_missing=None))
+ dst_port = Field(Int(min=0, max=65535, if_empty=None, if_missing=None))
+ def chained_validator(self, fields, state):
+ errors = dict()
+ if fields['protocol'] not in ('TCP', 'UDP'):
+ for name in ('src_port', 'dst_port'):
+ if fields[name] is not None:
+ errors[name] = u"Should be None if protocol " \
+ "(%(protocol)s) is not TCP or UDP" % fields
+ if errors:
+ raise Invalid(u"You can't specify any ports if the protocol "
+ u'is not TCP or UDP', fields, state, error_dict=errors)
+
+class RuleHandler(ListSubHandler):
+ r"""RuleHandler(parent) -> RuleHandler instance :: Handle a list of rules.
This class is a helper for FirewallHandler to do all the work related to rules
administration.
handler_help = u"Manage firewall rules"
- @handler(u'Add a new rule')
- def add(self, *args, **kwargs):
- r"add(rule) -> None :: Add a rule to the rules list (see Rule doc)."
- rule = Rule(*args, **kwargs)
- if rule in self.parent.rules:
- raise RuleAlreadyExistsError(rule)
- self.parent.rules.append(rule)
-
- @handler(u'Update a rule')
- def update(self, index, *args, **kwargs):
- r"update(index, rule) -> None :: Update a rule (see Rule doc)."
- # TODO check if the modified rule is the same of an existing one
- index = int(index) # TODO validation
- try:
- self.parent.rules[index].update(*args, **kwargs)
- except IndexError:
- raise RuleNotFoundError(index)
-
- @handler(u'Delete a rule')
- def delete(self, index):
- r"delete(index) -> Rule :: Delete a rule from the list returning it."
- index = int(index) # TODO validation
- try:
- return self.parent.rules.pop(index)
- except IndexError:
- raise RuleNotFoundError(index)
-
- @handler(u'Get information about a rule')
- def get(self, index):
- r"get(rule) -> Rule :: Get all the information about a rule."
- index = int(index) # TODO validation
- try:
- return self.parent.rules[index]
- except IndexError:
- raise RuleNotFoundError(index)
-
- @handler(u'Get information about all rules')
- def show(self):
- r"show() -> list of Rules :: List all the complete rules information."
- return self.parent.rules
-
+ _cont_subhandler_attr = 'rules'
+ _cont_subhandler_class = Rule
class FirewallHandler(Restorable, ConfigWriter, ServiceHandler,
TransactionalHandler):
handler_help = u"Manage firewall service"
- _persistent_attrs = 'rules'
+ _persistent_attrs = ['rules']
_restorable_defaults = dict(rules=list())
def __init__(self, pickle_dir='.', config_dir='.'):
r"Initialize the object, see class documentation for details."
+ log.debug(u'FirewallHandler(%r, %r)', pickle_dir, config_dir)
self._persistent_dir = pickle_dir
self._config_writer_cfg_dir = config_dir
self._service_start = ('sh', path.join(self._config_writer_cfg_dir,
self._service_restart = self._service_start
self._service_reload = self._service_start
self._config_build_templates()
- self._restore()
+ ServiceHandler.__init__(self)
self.rule = RuleHandler(self)
def _get_config_vars(self, config_file):
if __name__ == '__main__':
+ logging.basicConfig(
+ level = logging.DEBUG,
+ format = '%(asctime)s %(levelname)-8s %(message)s',
+ datefmt = '%H:%M:%S',
+ )
+
import os
fw_handler = FirewallHandler()
dump()
- fw_handler.rule.add('input','drop','icmp')
+ fw_handler.rule.add('input', 'drop', protocol='icmp')
fw_handler.rule.update(0, dst='192.168.0.188/32')
- fw_handler.rule.add('output','accept', '192.168.1.0/24')
+ fw_handler.rule.add('output', 'accept', '192.168.1.0/24')
fw_handler.commit()
projects / software / pymin.git / blobdiff