tiramisu/tiramisu/option/option.py

# -*- coding: utf-8 -*-
"option types and option description"
# Copyright (C) 2012-2017 Team tiramisu (see AUTHORS for all contributors)
#
# 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 3 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/>.
#
# The original `Config` design model is unproudly borrowed from
# the rough pypy's guys: http://codespeak.net/svn/pypy/dist/pypy/config/
# the whole pypy projet is under MIT licence
# ____________________________________________________________
import re
import sys
import datetime
from IPy import IP
from types import FunctionType
from ..setting import undefined

from ..error import ConfigError
from ..i18n import _
from .baseoption import Option, validate_callback, display_list
from ..autolib import carry_out_calculation


class ChoiceOption(Option):
    """represents a choice out of several objects.

    The option can also have the value ``None``
    """
    __slots__ = tuple()
    _display_name = _('choice')

    def __init__(self, name, doc, values, default=None,
                 values_params=None, default_multi=None, requires=None,
                 multi=False, callback=None, callback_params=None,
                 validator=None, validator_params=None,
                 properties=None, warnings_only=False):
        """
        :param values: is a list of values the option can possibly take
        """
        if isinstance(values, FunctionType):
            validate_callback(values, values_params, 'values', self)
        else:
            if values_params is not None:
                raise ValueError(_('values is not a function, so values_params must be None'))
            if not isinstance(values, tuple):
                raise TypeError(_('values must be a tuple or a function for {0}'
                                  ).format(name))
        self._choice_values = values
        if values_params is not None:
            self._choice_values_params = values_params
        super(ChoiceOption, self).__init__(name, doc, default=default,
                                           default_multi=default_multi,
                                           callback=callback,
                                           callback_params=callback_params,
                                           requires=requires,
                                           multi=multi,
                                           validator=validator,
                                           validator_params=validator_params,
                                           properties=properties,
                                           warnings_only=warnings_only)

    def impl_get_values(self, context, current_opt=undefined):
        if current_opt is undefined:
            current_opt = self
        #FIXME cache? but in context...
        values = self._choice_values
        if isinstance(values, FunctionType):
            if context is None:
                values = []
            else:
                values = carry_out_calculation(current_opt, context=context,
                                               callback=values,
                                               callback_params=getattr(self, '_choice_values_params', {}))
                if isinstance(values, Exception):
                    return values
                if values is not undefined and not isinstance(values, list):
                    raise ConfigError(_('calculated values for {0} is not a list'
                                        '').format(self.impl_getname()))
        return values


    def _validate(self, value, context=undefined, current_opt=undefined):
        values = self.impl_get_values(context, current_opt=current_opt)
        if isinstance(values, Exception):
            return values
        if values is not undefined and not value in values:
            if len(values) == 1:
                return ValueError(_('only {0} is allowed'
                                    '').format(values[0]))
            else:
                return ValueError(_('only {0} are allowed'
                                    '').format(display_list(values)))


class BoolOption(Option):
    "represents a choice between ``True`` and ``False``"
    __slots__ = tuple()
    _display_name = _('boolean')

    def _validate(self, value, context=undefined, current_opt=undefined):
        if not isinstance(value, bool):
            return ValueError()


class IntOption(Option):
    "represents a choice of an integer"
    __slots__ = tuple()
    _display_name = _('integer')

    def _validate(self, value, context=undefined, current_opt=undefined):
        if not isinstance(value, int):
            return ValueError()


class FloatOption(Option):
    "represents a choice of a floating point number"
    __slots__ = tuple()
    _display_name = _('float')

    def _validate(self, value, context=undefined, current_opt=undefined):
        if not isinstance(value, float):
            return ValueError()


class StrOption(Option):
    "represents the choice of a string"
    __slots__ = tuple()
    _display_name = _('string')

    def _validate(self, value, context=undefined, current_opt=undefined):
        if not isinstance(value, str):
            return ValueError()


if sys.version_info[0] >= 3:  # pragma: no cover
    #UnicodeOption is same as StrOption in python 3+
    class UnicodeOption(StrOption):
        __slots__ = tuple()
        pass
else:
    class UnicodeOption(Option):
        "represents the choice of a unicode string"
        __slots__ = tuple()
        _empty = u''
        _display_name = _('unicode string')

        def _validate(self, value, context=undefined, current_opt=undefined):
            if not isinstance(value, unicode):
                return ValueError()


class PasswordOption(Option):
    "represents the choice of a password"
    __slots__ = tuple()
    _display_name = _('password')

    def _validate(self, value, context=undefined, current_opt=undefined):
        err = self._impl_valid_unicode(value)
        if err:
            return err


class IPOption(Option):
    "represents the choice of an ip"
    __slots__ = tuple()
    _display_name = _('IP')

    def __init__(self, name, doc, default=None, default_multi=None,
                 requires=None, multi=False, callback=None,
                 callback_params=None, validator=None, validator_params=None,
                 properties=None, private_only=False, allow_reserved=False,
                 warnings_only=False):
        extra = {'_private_only': private_only,
                 '_allow_reserved': allow_reserved}
        super(IPOption, self).__init__(name, doc, default=default,
                                       default_multi=default_multi,
                                       callback=callback,
                                       callback_params=callback_params,
                                       requires=requires,
                                       multi=multi,
                                       validator=validator,
                                       validator_params=validator_params,
                                       properties=properties,
                                       warnings_only=warnings_only,
                                       extra=extra)

    def _validate(self, value, context=undefined, current_opt=undefined):
        # sometimes an ip term starts with a zero
        # but this does not fit in some case, for example bind does not like it
        err = self._impl_valid_unicode(value)
        if err:
            return err
        if value.count('.') != 3:
            return ValueError()
        for val in value.split('.'):
            if val.startswith("0") and len(val) > 1:
                return ValueError()
        # 'standard' validation
        try:
            IP('{0}/32'.format(value))
        except ValueError:
            return ValueError()

    def _second_level_validation(self, value, warnings_only):
        ip = IP('{0}/32'.format(value))
        if not self._get_extra('_allow_reserved') and ip.iptype() == 'RESERVED':
            if warnings_only:
                msg = _("shouldn't in reserved class")
            else:
                msg = _("mustn't be in reserved class")
            return ValueError(msg)
        if self._get_extra('_private_only') and not ip.iptype() == 'PRIVATE':
            if warnings_only:
                msg = _("should be in private class")
            else:
                msg = _("must be in private class")
            return ValueError(msg)

    def _cons_in_network(self, current_opt, opts, vals, warnings_only):
        if len(vals) != 3:
            raise ConfigError(_('invalid len for vals'))
        if None in vals:
            return
        ip, network, netmask = vals
        if IP(ip) not in IP('{0}/{1}'.format(network, netmask)):
            msg = _('{4} is not in network {0}/{1} ({2}/{3})')
            return ValueError(msg.format(network, netmask,
                              opts[1].impl_getname(), opts[2].impl_getname(), ip))
        # test if ip is not network/broadcast IP
        return opts[2]._cons_ip_netmask(current_opt, (opts[2], opts[0]), (netmask, ip), warnings_only)


class PortOption(Option):
    """represents the choice of a port
    The port numbers are divided into three ranges:
    the well-known ports,
    the registered ports,
    and the dynamic or private ports.
    You can actived this three range.
    Port number 0 is reserved and can't be used.
    see: http://en.wikipedia.org/wiki/Port_numbers
    """
    __slots__ = tuple()
    port_re = re.compile(r"^[0-9]*$")
    _display_name = _('port')

    def __init__(self, name, doc, default=None, default_multi=None,
                 requires=None, multi=False, callback=None,
                 callback_params=None, validator=None, validator_params=None,
                 properties=None, allow_range=False, allow_zero=False,
                 allow_wellknown=True, allow_registred=True,
                 allow_private=False, warnings_only=False):
        extra = {'_allow_range': allow_range,
                 '_min_value': None,
                 '_max_value': None}
        ports_min = [0, 1, 1024, 49152]
        ports_max = [0, 1023, 49151, 65535]
        is_finally = False
        for index, allowed in enumerate([allow_zero,
                                         allow_wellknown,
                                         allow_registred,
                                         allow_private]):
            if extra['_min_value'] is None:
                if allowed:
                    extra['_min_value'] = ports_min[index]
            elif not allowed:
                is_finally = True
            elif allowed and is_finally:
                raise ValueError(_('inconsistency in allowed range'))
            if allowed:
                extra['_max_value'] = ports_max[index]

        if extra['_max_value'] is None:
            raise ValueError(_('max value is empty'))

        super(PortOption, self).__init__(name, doc, default=default,
                                         default_multi=default_multi,
                                         callback=callback,
                                         callback_params=callback_params,
                                         requires=requires,
                                         multi=multi,
                                         validator=validator,
                                         validator_params=validator_params,
                                         properties=properties,
                                         warnings_only=warnings_only,
                                         extra=extra)

    def _validate(self, value, context=undefined, current_opt=undefined):
        if isinstance(value, int):
            if sys.version_info[0] >= 3:  # pragma: no cover
                value = str(value)
            else:
                value = unicode(value)
        err = self._impl_valid_unicode(value)
        if err:
            return err
        if self._get_extra('_allow_range') and ":" in str(value):
            value = str(value).split(':')
            if len(value) != 2:
                return ValueError(_('range must have two values only'))
            if not value[0] < value[1]:
                return ValueError(_('first port in range must be'
                                  ' smaller than the second one'))
        else:
            value = [value]

        for val in value:
            if not self.port_re.search(val):
                return ValueError()
            val = int(val)
            if not self._get_extra('_min_value') <= val <= self._get_extra('_max_value'):
                return ValueError(_('must be an integer between {0} '
                                    'and {1}').format(self._get_extra('_min_value'),
                                                      self._get_extra('_max_value')))


class NetworkOption(Option):
    "represents the choice of a network"
    __slots__ = tuple()
    _display_name = _('network address')

    def _validate(self, value, context=undefined, current_opt=undefined):
        err = self._impl_valid_unicode(value)
        if err:
            return err
        if value.count('.') != 3:
            return ValueError()
        for val in value.split('.'):
            if val.startswith("0") and len(val) > 1:
                return ValueError()
        try:
            IP(value)
        except ValueError:
            return ValueError()

    def _second_level_validation(self, value, warnings_only):
        ip = IP(value)
        if ip.iptype() == 'RESERVED':
            if warnings_only:
                msg = _("shouldn't be in reserved class")
            else:
                msg = _("mustn't be in reserved class")
            return ValueError(msg)


class NetmaskOption(Option):
    "represents the choice of a netmask"
    __slots__ = tuple()
    _display_name = _('netmask address')

    def _validate(self, value, context=undefined, current_opt=undefined):
        err = self._impl_valid_unicode(value)
        if err:
            return err
        if value.count('.') != 3:
            return ValueError()
        for val in value.split('.'):
            if val.startswith("0") and len(val) > 1:
                return ValueError()
        try:
            IP('0.0.0.0/{0}'.format(value))
        except ValueError:
            return ValueError()

    def _cons_network_netmask(self, current_opt, opts, vals, warnings_only):
        #opts must be (netmask, network) options
        if None in vals:
            return
        return self.__cons_netmask(opts, vals[0], vals[1], False, warnings_only)

    def _cons_ip_netmask(self, current_opt, opts, vals, warnings_only):
        #opts must be (netmask, ip) options
        if None in vals:
            return
        return self.__cons_netmask(opts, vals[0], vals[1], True, warnings_only)

    def __cons_netmask(self, opts, val_netmask, val_ipnetwork, make_net,
                       warnings_only):
        if len(opts) != 2:
            return ConfigError(_('invalid len for opts'))
        msg = None
        try:
            ip = IP('{0}/{1}'.format(val_ipnetwork, val_netmask),
                    make_net=make_net)
            #if cidr == 32, ip same has network
            if make_net and ip.prefixlen() != 32:
                val_ip = IP(val_ipnetwork)
                if ip.net() == val_ip:
                    msg = _("this is a network with netmask {0} ({1})")
                if ip.broadcast() == val_ip:
                    msg = _("this is a broadcast with netmask {0} ({1})")

        except ValueError:
            if not make_net:
                msg = _('with netmask {0} ({1})')
        if msg is not None:
            return ValueError(msg.format(val_netmask, opts[1].impl_getname()))


class BroadcastOption(Option):
    __slots__ = tuple()
    _display_name = _('broadcast address')

    def _validate(self, value, context=undefined, current_opt=undefined):
        err = self._impl_valid_unicode(value)
        if err:
            return err
        if value.count('.') != 3:
            return ValueError()
        for val in value.split('.'):
            if val.startswith("0") and len(val) > 1:
                return ValueError()
        try:
            IP('{0}/32'.format(value))
        except ValueError:
            return ValueError()

    def _cons_broadcast(self, current_opt, opts, vals, warnings_only):
        if len(vals) != 3:
            raise ConfigError(_('invalid len for vals'))
        if None in vals:
            return
        broadcast, network, netmask = vals
        if IP('{0}/{1}'.format(network, netmask)).broadcast() != IP(broadcast):
            return ValueError(_('broadcast {4} invalid with network {0}/{1} ({2}/{3})').format(
                network, netmask, opts[1].impl_getname(), opts[2].impl_getname(), broadcast))


class DomainnameOption(Option):
    """represents the choice of a domain name
    netbios: for MS domain
    hostname: to identify the device
    domainname:
    fqdn: with tld, not supported yet
    """
    __slots__ = tuple()
    _display_name = _('domain name')

    def __init__(self, name, doc, default=None, default_multi=None,
                 requires=None, multi=False, callback=None,
                 callback_params=None, validator=None, validator_params=None,
                 properties=None, allow_ip=False, type_='domainname',
                 warnings_only=False, allow_without_dot=False):
        if type_ not in ['netbios', 'hostname', 'domainname']:
            raise ValueError(_('unknown type_ {0} for hostname').format(type_))
        extra = {'_dom_type': type_}
        if allow_ip not in [True, False]:
            raise ValueError(_('allow_ip must be a boolean'))
        if allow_without_dot not in [True, False]:
            raise ValueError(_('allow_without_dot must be a boolean'))
        extra['_allow_ip'] = allow_ip
        extra['_allow_without_dot'] = allow_without_dot
        # FIXME should be
        # regexp = r'^((?!-)[a-z0-9-]{1,63}(?<!-)\.{0,1})$'
        if type_ == 'domainname':
            if allow_without_dot:
                min_time = 0
            else:
                min_time = 1
            regexp = r'((?!-)[a-z0-9-]{{{1},{0}}}\.){{{1},}}[a-z0-9-]{{1,{0}}}'.format(self._get_len(type_), min_time)
        else:
            regexp = r'((?!-)[a-z0-9-]{{1,{0}}})'.format(self._get_len(type_))
        if allow_ip:
            regexp = r'^(?:{0}|(?:(?:(?:25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)\.){{3}}(?:25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)))$'.format(regexp)
        else:
            regexp = r'^{0}$'.format(regexp)
        extra['_domain_re'] = re.compile(regexp)
        extra['_has_upper'] = re.compile('[A-Z]')

        super(DomainnameOption, self).__init__(name, doc, default=default,
                                               default_multi=default_multi,
                                               callback=callback,
                                               callback_params=callback_params,
                                               requires=requires,
                                               multi=multi,
                                               validator=validator,
                                               validator_params=validator_params,
                                               properties=properties,
                                               warnings_only=warnings_only,
                                               extra=extra)

    def _get_len(self, type_):
        if type_ == 'netbios':
            return 15
        else:
            return 63

    def _validate(self, value, context=undefined, current_opt=undefined):
        err = self._impl_valid_unicode(value)
        if err:
            return err

        def _valid_length(val):
            if len(val) < 1:
                return ValueError(_("invalid length (min 1)"))
            if len(val) > part_name_length:
                return ValueError(_("invalid length (max {0})"
                                    "").format(part_name_length))

        if self._get_extra('_allow_ip') is True:
            try:
                IP('{0}/32'.format(value))
                return
            except ValueError:
                pass
        else:
            try:
                IP('{0}/32'.format(value))
            except ValueError:
                pass
            else:
                return ValueError(_('must not be an IP'))
        part_name_length = self._get_len(self._get_extra('_dom_type'))
        if self._get_extra('_dom_type') == 'domainname':
            if not self._get_extra('_allow_without_dot') and not "." in value:
                return ValueError(_("must have dot"))
            if len(value) > 255:
                return ValueError(_("invalid length (max 255)"))
            for dom in value.split('.'):
                err = _valid_length(dom)
                if err:
                    return err
        else:
            return _valid_length(value)

    def _second_level_validation(self, value, warnings_only):
        if self._get_extra('_has_upper').search(value):
            return ValueError(_('some characters are uppercase'))
        if not self._get_extra('_domain_re').search(value):
            if warnings_only:
                return ValueError(_('some characters may cause problems'))
            else:
                return ValueError()


class URLOption(DomainnameOption):
    __slots__ = tuple()
    proto_re = re.compile(r'(http|https)://')
    path_re = re.compile(r"^[A-Za-z0-9\-\._~:/\?#\[\]@!%\$&\'\(\)\*\+,;=]+$")
    _display_name = _('URL')

    def _validate(self, value, context=undefined, current_opt=undefined):
        err = self._impl_valid_unicode(value)
        if err:
            return err
        match = self.proto_re.search(value)
        if not match:
            return ValueError(_('must start with http:// or '
                                'https://'))
        value = value[len(match.group(0)):]
        # get domain/files
        splitted = value.split('/', 1)
        if len(splitted) == 1:
            domain = value
            files = None
        else:
            domain, files = splitted
        # if port in domain
        splitted = domain.split(':', 1)
        if len(splitted) == 1:
            domain = splitted[0]
            port = 0
        else:
            domain, port = splitted
        if not 0 <= int(port) <= 65535:
            return ValueError(_('port must be an between 0 and '
                                '65536'))
        # validate domainname
        err = super(URLOption, self)._validate(domain)
        if err:
            return err
        err = super(URLOption, self)._second_level_validation(domain, False)
        if err:
            return err
        # validate file
        if files is not None and files != '' and not self.path_re.search(files):
            return ValueError(_('must ends with a valid resource name'))

    def _second_level_validation(self, value, warnings_only):
        pass


class _RegexpOption(Option):
    __slots__ = tuple()
    def _validate(self, value, context=undefined, current_opt=undefined):
        err = self._impl_valid_unicode(value)
        if err:
            return err
        match = self._regexp.search(value)
        if not match:
            return ValueError()


class EmailOption(_RegexpOption):
    __slots__ = tuple()
    #https://www.w3.org/TR/html-markup/input.email.html#input.email.attrs.value.single.
    _regexp = re.compile(r"^[a-zA-Z0-9.!#$%&'*+/=?^_`{|}~-]+@[a-zA-Z0-9-]+(?:\.[a-zA-Z0-9-]+)*$")
    _display_name = _('email address')


class UsernameOption(_RegexpOption):
    __slots__ = tuple()
    #regexp build with 'man 8 adduser' informations
    _regexp = re.compile(r"^[a-z_][a-z0-9_-]{0,30}[$a-z0-9_-]{0,1}$")
    _display_name = _('username')


class FilenameOption(_RegexpOption):
    __slots__ = tuple()
    _regexp = re.compile(r"^[a-zA-Z0-9\-\._~/+]+$")
    _display_name = _('file name')


class DateOption(Option):
    __slots__ = tuple()
    _display_name = _('date')

    def _validate(self, value, context=undefined, current_opt=undefined):
        err = self._impl_valid_unicode(value)
        if err:
            return err
        try:
            datetime.datetime.strptime(value,"%Y-%m-%d")
        except ValueError:
            return ValueError()