source: uri/rfc3986/validators.py@ 1292

# -*- coding: utf-8 -*-
# Copyright (c) 2017 Ian Stapleton Cordasco
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#    http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
# implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""Module containing the validation logic for rfc3986."""
from . import exceptions
from . import misc
from . import normalizers


class Validator(object):
    """Object used to configure validation of all objects in rfc3986.

    .. versionadded:: 1.0

    Example usage::

         >>> from rfc3986 import api, validators
         >>> uri = api.uri_reference('https://github.com/')
         >>> validator = validators.Validator().require_presence_of(
         ...    'scheme', 'host', 'path',
         ... ).allow_schemes(
         ...    'http', 'https',
         ... ).allow_hosts(
         ...    '127.0.0.1', 'github.com',
         ... )
         >>> validator.validate(uri)
         >>> invalid_uri = rfc3986.uri_reference('imap://mail.google.com')
         >>> validator.validate(invalid_uri)
         Traceback (most recent call last):
         ...
         rfc3986.exceptions.MissingComponentError: ('path was required but
         missing', URIReference(scheme=u'imap', authority=u'mail.google.com',
         path=None, query=None, fragment=None), ['path'])

    """

    COMPONENT_NAMES = frozenset(
        ["scheme", "userinfo", "host", "port", "path", "query", "fragment"]
    )

    def __init__(self):
        """Initialize our default validations."""
        self.allowed_schemes = set()
        self.allowed_hosts = set()
        self.allowed_ports = set()
        self.allow_password = True
        self.required_components = {
            "scheme": False,
            "userinfo": False,
            "host": False,
            "port": False,
            "path": False,
            "query": False,
            "fragment": False,
        }
        self.validated_components = self.required_components.copy()

    def allow_schemes(self, *schemes):
        """Require the scheme to be one of the provided schemes.

        .. versionadded:: 1.0

        :param schemes:
            Schemes, without ``://`` that are allowed.
        :returns:
            The validator instance.
        :rtype:
            Validator
        """
        for scheme in schemes:
            self.allowed_schemes.add(normalizers.normalize_scheme(scheme))
        return self

    def allow_hosts(self, *hosts):
        """Require the host to be one of the provided hosts.

        .. versionadded:: 1.0

        :param hosts:
            Hosts that are allowed.
        :returns:
            The validator instance.
        :rtype:
            Validator
        """
        for host in hosts:
            self.allowed_hosts.add(normalizers.normalize_host(host))
        return self

    def allow_ports(self, *ports):
        """Require the port to be one of the provided ports.

        .. versionadded:: 1.0

        :param ports:
            Ports that are allowed.
        :returns:
            The validator instance.
        :rtype:
            Validator
        """
        for port in ports:
            port_int = int(port, base=10)
            if 0 <= port_int <= 65535:
                self.allowed_ports.add(port)
        return self

    def allow_use_of_password(self):
        """Allow passwords to be present in the URI.

        .. versionadded:: 1.0

        :returns:
            The validator instance.
        :rtype:
            Validator
        """
        self.allow_password = True
        return self

    def forbid_use_of_password(self):
        """Prevent passwords from being included in the URI.

        .. versionadded:: 1.0

        :returns:
            The validator instance.
        :rtype:
            Validator
        """
        self.allow_password = False
        return self

    def check_validity_of(self, *components):
        """Check the validity of the components provided.

        This can be specified repeatedly.

        .. versionadded:: 1.1

        :param components:
            Names of components from :attr:`Validator.COMPONENT_NAMES`.
        :returns:
            The validator instance.
        :rtype:
            Validator
        """
        components = [c.lower() for c in components]
        for component in components:
            if component not in self.COMPONENT_NAMES:
                raise ValueError(
                    '"{}" is not a valid component'.format(component)
                )
        self.validated_components.update(
            {component: True for component in components}
        )
        return self

    def require_presence_of(self, *components):
        """Require the components provided.

        This can be specified repeatedly.

        .. versionadded:: 1.0

        :param components:
            Names of components from :attr:`Validator.COMPONENT_NAMES`.
        :returns:
            The validator instance.
        :rtype:
            Validator
        """
        components = [c.lower() for c in components]
        for component in components:
            if component not in self.COMPONENT_NAMES:
                raise ValueError(
                    '"{}" is not a valid component'.format(component)
                )
        self.required_components.update(
            {component: True for component in components}
        )
        return self

    def validate(self, uri):
        """Check a URI for conditions specified on this validator.

        .. versionadded:: 1.0

        :param uri:
            Parsed URI to validate.
        :type uri:
            rfc3986.uri.URIReference
        :raises MissingComponentError:
            When a required component is missing.
        :raises UnpermittedComponentError:
            When a component is not one of those allowed.
        :raises PasswordForbidden:
            When a password is present in the userinfo component but is
            not permitted by configuration.
        :raises InvalidComponentsError:
            When a component was found to be invalid.
        """
        if not self.allow_password:
            check_password(uri)

        required_components = [
            component
            for component, required in self.required_components.items()
            if required
        ]
        validated_components = [
            component
            for component, required in self.validated_components.items()
            if required
        ]
        if required_components:
            ensure_required_components_exist(uri, required_components)
        if validated_components:
            ensure_components_are_valid(uri, validated_components)

        ensure_one_of(self.allowed_schemes, uri, "scheme")
        ensure_one_of(self.allowed_hosts, uri, "host")
        ensure_one_of(self.allowed_ports, uri, "port")


def check_password(uri):
    """Assert that there is no password present in the uri."""
    userinfo = uri.userinfo
    if not userinfo:
        return
    credentials = userinfo.split(":", 1)
    if len(credentials) <= 1:
        return
    raise exceptions.PasswordForbidden(uri)


def ensure_one_of(allowed_values, uri, attribute):
    """Assert that the uri's attribute is one of the allowed values."""
    value = getattr(uri, attribute)
    if value is not None and allowed_values and value not in allowed_values:
        raise exceptions.UnpermittedComponentError(
            attribute,
            value,
            allowed_values,
        )


def ensure_required_components_exist(uri, required_components):
    """Assert that all required components are present in the URI."""
    missing_components = sorted(
        [
            component
            for component in required_components
            if getattr(uri, component) is None
        ]
    )
    if missing_components:
        raise exceptions.MissingComponentError(uri, *missing_components)


def is_valid(value, matcher, require):
    """Determine if a value is valid based on the provided matcher.

    :param str value:
        Value to validate.
    :param matcher:
        Compiled regular expression to use to validate the value.
    :param require:
        Whether or not the value is required.
    """
    if require:
        return value is not None and matcher.match(value)

    # require is False and value is not None
    return value is None or matcher.match(value)


def authority_is_valid(authority, host=None, require=False):
    """Determine if the authority string is valid.

    :param str authority:
        The authority to validate.
    :param str host:
        (optional) The host portion of the authority to validate.
    :param bool require:
        (optional) Specify if authority must not be None.
    :returns:
        ``True`` if valid, ``False`` otherwise
    :rtype:
        bool
    """
    validated = is_valid(authority, misc.SUBAUTHORITY_MATCHER, require)
    if validated and host is not None:
        return host_is_valid(host, require)
    return validated


def host_is_valid(host, require=False):
    """Determine if the host string is valid.

    :param str host:
        The host to validate.
    :param bool require:
        (optional) Specify if host must not be None.
    :returns:
        ``True`` if valid, ``False`` otherwise
    :rtype:
        bool
    """
    validated = is_valid(host, misc.HOST_MATCHER, require)
    if validated and host is not None and misc.IPv4_MATCHER.match(host):
        return valid_ipv4_host_address(host)
    elif validated and host is not None and misc.IPv6_MATCHER.match(host):
        return misc.IPv6_NO_RFC4007_MATCHER.match(host) is not None
    return validated


def scheme_is_valid(scheme, require=False):
    """Determine if the scheme is valid.

    :param str scheme:
        The scheme string to validate.
    :param bool require:
        (optional) Set to ``True`` to require the presence of a scheme.
    :returns:
        ``True`` if the scheme is valid. ``False`` otherwise.
    :rtype:
        bool
    """
    return is_valid(scheme, misc.SCHEME_MATCHER, require)


def path_is_valid(path, require=False):
    """Determine if the path component is valid.

    :param str path:
        The path string to validate.
    :param bool require:
        (optional) Set to ``True`` to require the presence of a path.
    :returns:
        ``True`` if the path is valid. ``False`` otherwise.
    :rtype:
        bool
    """
    return is_valid(path, misc.PATH_MATCHER, require)


def query_is_valid(query, require=False):
    """Determine if the query component is valid.

    :param str query:
        The query string to validate.
    :param bool require:
        (optional) Set to ``True`` to require the presence of a query.
    :returns:
        ``True`` if the query is valid. ``False`` otherwise.
    :rtype:
        bool
    """
    return is_valid(query, misc.QUERY_MATCHER, require)


def fragment_is_valid(fragment, require=False):
    """Determine if the fragment component is valid.

    :param str fragment:
        The fragment string to validate.
    :param bool require:
        (optional) Set to ``True`` to require the presence of a fragment.
    :returns:
        ``True`` if the fragment is valid. ``False`` otherwise.
    :rtype:
        bool
    """
    return is_valid(fragment, misc.FRAGMENT_MATCHER, require)


def valid_ipv4_host_address(host):
    """Determine if the given host is a valid IPv4 address."""
    # If the host exists, and it might be IPv4, check each byte in the
    # address.
    return all([0 <= int(byte, base=10) <= 255 for byte in host.split(".")])


_COMPONENT_VALIDATORS = {
    "scheme": scheme_is_valid,
    "path": path_is_valid,
    "query": query_is_valid,
    "fragment": fragment_is_valid,
}

_SUBAUTHORITY_VALIDATORS = set(["userinfo", "host", "port"])


def subauthority_component_is_valid(uri, component):
    """Determine if the userinfo, host, and port are valid."""
    try:
        subauthority_dict = uri.authority_info()
    except exceptions.InvalidAuthority:
        return False

    # If we can parse the authority into sub-components and we're not
    # validating the port, we can assume it's valid.
    if component == "host":
        return host_is_valid(subauthority_dict["host"])
    elif component != "port":
        return True

    try:
        port = int(subauthority_dict["port"])
    except TypeError:
        # If the port wasn't provided it'll be None and int(None) raises a
        # TypeError
        return True

    return 0 <= port <= 65535


def ensure_components_are_valid(uri, validated_components):
    """Assert that all components are valid in the URI."""
    invalid_components = set([])
    for component in validated_components:
        if component in _SUBAUTHORITY_VALIDATORS:
            if not subauthority_component_is_valid(uri, component):
                invalid_components.add(component)
            # Python's peephole optimizer means that while this continue *is*
            # actually executed, coverage.py cannot detect that. See also,
            # https://bitbucket.org/ned/coveragepy/issues/198/continue-marked-as-not-covered
            continue  # nocov: Python 2.7, 3.3, 3.4

        validator = _COMPONENT_VALIDATORS[component]
        if not validator(getattr(uri, component)):
            invalid_components.add(component)

    if invalid_components:
        raise exceptions.InvalidComponentsError(uri, *invalid_components)