docker-compose/compose/config/validation.py

import json
import logging
import os
import sys

import six
from docker.utils.ports import split_port
from jsonschema import Draft4Validator
from jsonschema import FormatChecker
from jsonschema import RefResolver
from jsonschema import ValidationError

from .errors import ConfigurationError


log = logging.getLogger(__name__)


DOCKER_CONFIG_HINTS = {
    'cpu_share': 'cpu_shares',
    'add_host': 'extra_hosts',
    'hosts': 'extra_hosts',
    'extra_host': 'extra_hosts',
    'device': 'devices',
    'link': 'links',
    'memory_swap': 'memswap_limit',
    'port': 'ports',
    'privilege': 'privileged',
    'priviliged': 'privileged',
    'privilige': 'privileged',
    'volume': 'volumes',
    'workdir': 'working_dir',
}


VALID_NAME_CHARS = '[a-zA-Z0-9\._\-]'


@FormatChecker.cls_checks(
    format="ports",
    raises=ValidationError(
        "Invalid port formatting, it should be "
        "'[[remote_ip:]remote_port:]port[/protocol]'"))
def format_ports(instance):
    try:
        split_port(instance)
    except ValueError:
        return False
    return True


@FormatChecker.cls_checks(format="environment")
def format_boolean_in_environment(instance):
    """
    Check if there is a boolean in the environment and display a warning.
    Always return True here so the validation won't raise an error.
    """
    if isinstance(instance, bool):
        log.warn(
            "There is a boolean value in the 'environment' key.\n"
            "Environment variables can only be strings.\n"
            "Please add quotes to any boolean values to make them string "
            "(eg, 'True', 'yes', 'N').\n"
            "This warning will become an error in a future release. \r\n"
        )
    return True


def validate_service_names(config):
    for service_name in config.keys():
        if not isinstance(service_name, six.string_types):
            raise ConfigurationError(
                "Service name: {} needs to be a string, eg '{}'".format(
                    service_name,
                    service_name))


def validate_top_level_object(config):
    if not isinstance(config, dict):
        raise ConfigurationError(
            "Top level object needs to be a dictionary. Check your .yml file "
            "that you have defined a service at the top level.")
    validate_service_names(config)


def validate_extends_file_path(service_name, extends_options, filename):
    """
    The service to be extended must either be defined in the config key 'file',
    or within 'filename'.
    """
    error_prefix = "Invalid 'extends' configuration for %s:" % service_name

    if 'file' not in extends_options and filename is None:
        raise ConfigurationError(
            "%s you need to specify a 'file', e.g. 'file: something.yml'" % error_prefix
        )


def get_unsupported_config_msg(service_name, error_key):
    msg = "Unsupported config option for '{}' service: '{}'".format(service_name, error_key)
    if error_key in DOCKER_CONFIG_HINTS:
        msg += " (did you mean '{}'?)".format(DOCKER_CONFIG_HINTS[error_key])
    return msg


def anglicize_validator(validator):
    if validator in ["array", "object"]:
        return 'an ' + validator
    return 'a ' + validator


def handle_error_for_schema_with_id(error, service_name):
    schema_id = error.schema['id']

    if schema_id == 'fields_schema.json' and error.validator == 'additionalProperties':
        return "Invalid service name '{}' - only {} characters are allowed".format(
            # The service_name is the key to the json object
            list(error.instance)[0],
            VALID_NAME_CHARS)

    if schema_id == '#/definitions/constraints':
        if 'image' in error.instance and 'build' in error.instance:
            return (
                "Service '{}' has both an image and build path specified. "
                "A service can either be built to image or use an existing "
                "image, not both.".format(service_name))
        if 'image' not in error.instance and 'build' not in error.instance:
            return (
                "Service '{}' has neither an image nor a build path "
                "specified. Exactly one must be provided.".format(service_name))
        if 'image' in error.instance and 'dockerfile' in error.instance:
            return (
                "Service '{}' has both an image and alternate Dockerfile. "
                "A service can either be built to image or use an existing "
                "image, not both.".format(service_name))

    if schema_id == '#/definitions/service':
        if error.validator == 'additionalProperties':
            invalid_config_key = parse_key_from_error_msg(error)
            return get_unsupported_config_msg(service_name, invalid_config_key)


def handle_generic_service_error(error, service_name):
    config_key = " ".join("'%s'" % k for k in error.path)
    msg_format = None
    error_msg = error.message

    if error.validator == 'oneOf':
        msg_format = "Service '{}' configuration key {} {}"
        error_msg = _parse_oneof_validator(error)

    elif error.validator == 'type':
        msg_format = ("Service '{}' configuration key {} contains an invalid "
                      "type, it should be {}")
        error_msg = _parse_valid_types_from_validator(error.validator_value)

    # TODO: no test case for this branch, there are no config options
    # which exercise this branch
    elif error.validator == 'required':
        msg_format = "Service '{}' configuration key '{}' is invalid, {}"

    elif error.validator == 'dependencies':
        msg_format = "Service '{}' configuration key '{}' is invalid: {}"
        config_key = list(error.validator_value.keys())[0]
        required_keys = ",".join(error.validator_value[config_key])
        error_msg = "when defining '{}' you must set '{}' as well".format(
            config_key,
            required_keys)

    elif error.path:
        msg_format = "Service '{}' configuration key {} value {}"

    if msg_format:
        return msg_format.format(service_name, config_key, error_msg)

    return error.message


def parse_key_from_error_msg(error):
    return error.message.split("'")[1]


def _parse_valid_types_from_validator(validator):
    """A validator value can be either an array of valid types or a string of
    a valid type. Parse the valid types and prefix with the correct article.
    """
    if not isinstance(validator, list):
        return anglicize_validator(validator)

    if len(validator) == 1:
        return anglicize_validator(validator[0])

    return "{}, or {}".format(
        ", ".join([anglicize_validator(validator[0])] + validator[1:-1]),
        anglicize_validator(validator[-1]))


def _parse_oneof_validator(error):
    """oneOf has multiple schemas, so we need to reason about which schema, sub
    schema or constraint the validation is failing on.
    Inspecting the context value of a ValidationError gives us information about
    which sub schema failed and which kind of error it is.
    """
    types = []
    for context in error.context:

        if context.validator == 'required':
            return context.message

        if context.validator == 'additionalProperties':
            invalid_config_key = parse_key_from_error_msg(context)
            return "contains unsupported option: '{}'".format(invalid_config_key)

        if context.path:
            invalid_config_key = " ".join(
                "'{}' ".format(fragment) for fragment in context.path
                if isinstance(fragment, six.string_types)
            )
            return "{}contains {}, which is an invalid type, it should be {}".format(
                invalid_config_key,
                context.instance,
                _parse_valid_types_from_validator(context.validator_value))

        if context.validator == 'uniqueItems':
            return "contains non unique items, please remove duplicates from {}".format(
                context.instance)

        if context.validator == 'type':
            types.append(context.validator_value)

    valid_types = _parse_valid_types_from_validator(types)
    return "contains an invalid type, it should be {}".format(valid_types)


def process_errors(errors, service_name=None):
    """jsonschema gives us an error tree full of information to explain what has
    gone wrong. Process each error and pull out relevant information and re-write
    helpful error messages that are relevant.
    """
    def format_error_message(error, service_name):
        if not service_name and error.path:
            # field_schema errors will have service name on the path
            service_name = error.path.popleft()

        if 'id' in error.schema:
            error_msg = handle_error_for_schema_with_id(error, service_name)
            if error_msg:
                return error_msg

        return handle_generic_service_error(error, service_name)

    return '\n'.join(format_error_message(error, service_name) for error in errors)


def validate_against_fields_schema(config):
    return _validate_against_schema(
        config,
        "fields_schema.json",
        ["ports", "environment"])


def validate_against_service_schema(config, service_name):
    return _validate_against_schema(
        config,
        "service_schema.json",
        ["ports"],
        service_name)


def _validate_against_schema(
        config,
        schema_filename,
        format_checker=(),
        service_name=None):
    config_source_dir = os.path.dirname(os.path.abspath(__file__))

    if sys.platform == "win32":
        file_pre_fix = "///"
        config_source_dir = config_source_dir.replace('\\', '/')
    else:
        file_pre_fix = "//"

    resolver_full_path = "file:{}{}/".format(file_pre_fix, config_source_dir)
    schema_file = os.path.join(config_source_dir, schema_filename)

    with open(schema_file, "r") as schema_fh:
        schema = json.load(schema_fh)

    resolver = RefResolver(resolver_full_path, schema)
    validation_output = Draft4Validator(
        schema,
        resolver=resolver,
        format_checker=FormatChecker(format_checker))

    errors = [error for error in sorted(validation_output.iter_errors(config), key=str)]
    if errors:
        error_msg = process_errors(errors, service_name)
        raise ConfigurationError("Validation failed, reason(s):\n{}".format(error_msg))