api/validate.py

# Copyright 2019 The Chromium OS Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.

"""Validation helpers for simple input validation in the API.

Note: Every validator MUST respect config.do_validation. This is an internally
set config option that allows the mock call decorators to be placed before or
after the validation decorators, rather than forcing an ordering that could then
produce incorrect outputs if missed.
"""

import functools
import logging
import os
from typing import Callable, Iterable, List, Optional, Union

from chromite.third_party.google.protobuf import message as protobuf_message

from chromite.lib import cros_build_lib


def _value(
    field: str, message: protobuf_message.Message
) -> Union[bool, int, str, None, List, protobuf_message.Message]:
    """Helper function to fetch the value of the field.

    Args:
      field: The field name. Can be nested via . separation.
      message: The protobuf message it is being fetched from.

    Returns:
      The value of the field.
    """
    if not field:
        return message

    value = message
    for part in field.split("."):
        if not isinstance(value, protobuf_message.Message):
            value = None
            break

        try:
            value = getattr(value, part)
        except AttributeError as e:
            cros_build_lib.Die("Invalid field: %s", e)

    return value


# pylint: disable=docstring-misnamed-args
def exists(*fields: str):
    """Validate that the paths in |fields| exist.

    Args:
      fields (str): The fields being checked. Can be . separated nested
        fields.
    """
    assert fields

    def decorator(func):
        @functools.wraps(func)
        def _exists(input_proto, output_proto, config, *args, **kwargs):
            if config.do_validation:
                for field in fields:
                    logging.debug("Validating %s exists.", field)

                    value = _value(field, input_proto)
                    if not value or not os.path.exists(value):
                        cros_build_lib.Die(
                            "%s path does not exist: %s" % (field, value)
                        )

            return func(input_proto, output_proto, config, *args, **kwargs)

        return _exists

    return decorator


def is_in(field: str, values: Iterable):
    """Validate |field| is an element of |values|.

    Args:
      field: The field being checked. May be . separated nested fields.
      values: The possible values field may take.
    """
    assert field
    assert values

    def decorator(func):
        @functools.wraps(func)
        def _is_in(input_proto, output_proto, config, *args, **kwargs):
            if config.do_validation:
                logging.debug("Validating %s is in %r", field, values)
                value = _value(field, input_proto)

                if value not in values:
                    cros_build_lib.Die(
                        "%s (%r) must be in %r", field, value, values
                    )

            return func(input_proto, output_proto, config, *args, **kwargs)

        return _is_in

    return decorator


def each_in(
    field: str,
    subfield: Optional[str],
    values: Iterable,
    optional: bool = False,
):
    """Validate each |subfield| of the repeated |field| is in |values|.

    Args:
      field: The field being checked. May be . separated nested fields.
      subfield: The field in the repeated |field| to validate, or None
        when |field| is not a repeated message, e.g. enum, scalars.
      values: The possible values field may take.
      optional: Also allow the field to be empty when True.
    """
    assert field
    assert values

    def decorator(func):
        @functools.wraps(func)
        def _is_in(input_proto, output_proto, config, *args, **kwargs):
            if config.do_validation:
                members = _value(field, input_proto) or []
                if not optional and not members:
                    cros_build_lib.Die("The %s field is empty.", field)
                for member in members:
                    logging.debug(
                        "Validating %s.[each].%s is in %r.",
                        field,
                        subfield,
                        values,
                    )
                    value = _value(subfield, member)
                    if value not in values:
                        cros_build_lib.Die(
                            "%s.[each].%s (%r) must be in %r is required.",
                            field,
                            subfield,
                            value,
                            values,
                        )

            return func(input_proto, output_proto, config, *args, **kwargs)

        return _is_in

    return decorator


def constraint(description):
    """Define a function to be used as a constraint check.

    A constraint is a function that checks the value of a field and either
    does nothing (returns None) or returns a string indicating why the value
    isn't valid.

    We bind a human readable description to the constraint for error reporting
    and logging.

    Args:
      description: Human readable description of the constraint
    """

    def decorator(func):
        @functools.wraps(func)
        def _func(*args, **kwargs):
            func(*args, **kwargs)

        setattr(_func, "__constraint_description__", description)
        return _func

    return decorator


def check_constraint(field: str, checkfunc: Callable):
    """Validate all values of |field| pass a constraint.

    Args:
      field: The field being checked. May be . separated nested fields.
      checkfunc: A constraint function to check on each value
    """
    assert field
    assert constraint

    # Get description for the constraint if it's set
    constraint_description = getattr(
        checkfunc,
        "__constraint_description__",
        checkfunc.__name__,
    )

    def decorator(func):
        @functools.wraps(func)
        def _check_constraint(
            input_proto, output_proto, config, *args, **kwargs
        ):
            if config.do_validation:
                values = _value(field, input_proto) or []

                failed = []
                for val in values:
                    msg = checkfunc(val)
                    if msg is not None:
                        failed.append((val, msg))

                if failed:
                    msg = (
                        f"{field}.[all] one or more values failed check "
                        f'"{constraint_description}"\n'
                    )

                    for value, msg in failed:
                        msg += "  %s: %s\n" % (value, msg)
                    cros_build_lib.Die(msg)

            return func(input_proto, output_proto, config, *args, **kwargs)

        return _check_constraint

    return decorator


# pylint: disable=docstring-misnamed-args
def require(*fields: str):
    """Verify |fields| have all been set to truthy values.

    Args:
      fields: The fields being checked. May be . separated nested fields.
    """
    assert fields

    def decorator(func):
        @functools.wraps(func)
        def _require(input_proto, output_proto, config, *args, **kwargs):
            if config.do_validation:
                for field in fields:
                    logging.debug("Validating %s is set.", field)

                    value = _value(field, input_proto)
                    if not value:
                        cros_build_lib.Die("%s is required.", field)

            return func(input_proto, output_proto, config, *args, **kwargs)

        return _require

    return decorator


# pylint: disable=docstring-misnamed-args
def require_any(*fields: str):
    """Verify at least one of |fields| have been set.

    Args:
      fields: The fields being checked. May be . separated nested fields.
    """
    assert fields

    def decorator(func):
        @functools.wraps(func)
        def _require(input_proto, output_proto, config, *args, **kwargs):
            if config.do_validation:
                for field in fields:
                    logging.debug("Validating %s is set.", field)
                    value = _value(field, input_proto)
                    if value:
                        break
                else:
                    cros_build_lib.Die(
                        "At least one of the following must be set: %s",
                        ", ".join(fields),
                    )

            return func(input_proto, output_proto, config, *args, **kwargs)

        return _require

    return decorator


def require_each(
    field: str, subfields: Iterable[str], allow_empty: bool = True
):
    """Verify |field| each have all of the |subfields| set.

    When |allow_empty| is True, |field| may be empty, and |subfields| are only
    validated when it is not empty. When |allow_empty| is False, |field| must
    also have at least one entry.

    Args:
      field: The repeated field being checked. May be . separated nested
          fields.
      subfields: The fields of the repeated message to validate.
      allow_empty: Also require at least one entry in the repeated field.
    """
    assert field
    assert subfields
    assert not isinstance(subfields, str)

    def decorator(func):
        @functools.wraps(func)
        def _require_each(input_proto, output_proto, config, *args, **kwargs):
            if config.do_validation:
                members = _value(field, input_proto) or []
                if not allow_empty and not members:
                    cros_build_lib.Die("The %s field is empty.", field)
                for member in members:
                    for subfield in subfields:
                        logging.debug(
                            "Validating %s.[each].%s is set.", field, subfield
                        )
                        value = _value(subfield, member)
                        if not value:
                            cros_build_lib.Die("%s is required.", field)

            return func(input_proto, output_proto, config, *args, **kwargs)

        return _require_each

    return decorator


def validation_complete(func: Callable):
    """Automatically skip the endpoint when called after all other validators.

    This decorator MUST be applied after all other validate decorators.
    The config can be checked manually if there is non-decorator validation, but
    this is much cleaner if it is all done in decorators.
    """

    @functools.wraps(func)
    def _validate_only(request, response, configs, *args, **kwargs):
        if configs.validate_only:
            # Avoid calling the endpoint.
            return 0
        else:
            return func(request, response, configs, *args, **kwargs)

    return _validate_only