lib/xos-genx/xosgenx/validator.py

# Copyright 2017-present Open Networking Foundation
#
# 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.

"""
  This module is used to validate xproto models and fields. The basic guiding principle is everything that isn't
  specifically allowed here should be denied by default.

  Note: While xproto must maintain some compatibility with django give the implementation choice of using django
  in the core, it's the case that the allowable set of xproto options may be a subset of what is allowed under
  django. For example, there may be django features that do not need exposure in xproto and/or are incompatible
  with other design aspects of XOS such as the XOS gRPC API  implementation.
"""

from __future__ import print_function
import sys
import os


# Options that are always allowed
COMMON_OPTIONS = ["help_text", "gui_hidden", "tosca_key", "tosca_key_one_of",
                  "feedback_state", "unique", "unique_with"]

# Options that must be either "True" or "False"
BOOLEAN_OPTIONS = ["blank", "db_index", "feedback_state", "gui_hidden", "null", "tosca_key", "unique", "varchar"]


class XProtoValidator(object):
    def __init__(self, models, line_map):
        """
            models: a list of model definitions. Each model is a dictionary.
            line_map: a list of tuples (start_line_no, filename) that tells which file goes with which line number.
        """
        self.models = models
        self.line_map = line_map
        self.errors = []

    def error(self, model, field, message):
        if field and field.get("_linespan"):
            error_first_line_number = field["_linespan"][0]
            error_last_line_number = field["_linespan"][1]
        else:
            error_first_line_number = model["_linespan"][0]
            error_last_line_number = model["_linespan"][1]

        error_filename = "unknown"
        error_line_offset = 0
        for (start_line, fn) in self.line_map:
            if start_line > error_first_line_number:
                break
            error_filename = fn
            error_line_offset = start_line

        self.errors.append({"model": model,
                            "field": field,
                            "message": message,
                            "filename": error_filename,
                            "first_line_number": error_first_line_number - error_line_offset,
                            "last_line_number": error_last_line_number - error_line_offset,
                            "absolute_line_number": error_first_line_number})

    def print_errors(self):
        # Sort by line number
        for error in sorted(self.errors, key=lambda error: error["absolute_line_number"]):
            model = error["model"]
            field = error["field"]
            message = error["message"]
            first_line_number = error["first_line_number"]
            last_line_number = error["last_line_number"]

            if first_line_number != last_line_number:
                linestr = "%d-%d" % (first_line_number, last_line_number)
            else:
                linestr = "%d" % first_line_number

            print("[ERROR] %s:%s %s.%s (Type %s): %s" % (os.path.basename(error["filename"]),
                                                         linestr,
                                                         model.get("name"),
                                                         field.get("name"),
                                                         field.get("type"),
                                                         message), file=sys.stderr)

    def is_option_true(self, field, name):
        options = field.get("options")
        if not options:
            return False
        option = options.get(name)
        return option == "True"

    def allow_options(self, model, field, options):
        """ Only allow the options specified in `options`. If some option is present that isn't in allowed, then
            register an error.

            `options` is a list of options which can either be simple names, or `name=value`.
        """
        options = COMMON_OPTIONS + options

        for (k, v) in field.get("options", {}).items():
            allowed = False
            for option in options:
                if "=" in option:
                    (optname, optval) = option.split("=")
                    if optname == k and optval == v:
                        allowed = True
                else:
                    if option == k:
                        allowed = True

            if not allowed:
                self.error(model, field, "Option %s=%s is not allowed" % (k, v))

            if k in BOOLEAN_OPTIONS and (v not in ["True", "False"]):
                self.error(model, field, "Option `%s` must be either True or False, but is '%s'" % (k, v))

    def require_options(self, model, field, options):
        """ Require an option to be present.
        """
        options = field.get("options", {})
        for optname in options:
            if optname not in options:
                self.error(model, field, "Required option '%s' is not present" % optname)

    def check_modifier_consistent(self, model, field):
        """ Validates that "modifier" is consistent with options.

            Required/optional imply some settings for blank= and null=. These settings are dependent on the type
            of field. See also jinja2_extensions/django.py which has to implement some of the same logic.
        """
        field_type = field["type"]
        options = field.get("options", {})
        modifier = options.get('modifier')
        link_type = field.get("link_type")
        mod_out = {}

        if modifier == "required":
            mod_out["blank"] = 'False'

            if link_type != "manytomany":
                mod_out["null"] = 'False'

        elif modifier == "optional":
            mod_out["blank"] = 'True'

            # set defaults on link types
            if link_type != "manytomany" and field_type != "bool":
                mod_out["null"] = 'True'

        else:
            self.error(model, field, "Unknown modifier type '%s'" % modifier)

        # print an error if there's a field conflict
        for kmo in mod_out.keys():
            if (kmo in options) and (options[kmo] != mod_out[kmo]):
                self.error(model, field, "Option `%s`=`%s` is inconsistent with modifier `%s`" %
                           (kmo, options[kmo], modifier))

    def validate_field_date(self, model, field):
        self.check_modifier_consistent(model, field)
        self.allow_options(model, field,
                           ["auto_now_add", "blank", "db_index", "default",
                            "max_length", "modifier", "null", "content_type"])

    def validate_field_string(self, model, field):
        # A string with a `content_type="date"` is actually a date
        # TODO: Investigate why there are double-quotes around "date"
        content_type = field.get("options", {}).get("content_type")
        if content_type in ["\"date\""]:
            self.validate_field_date(model, field)
            return

        # TODO: Investigate why there are double-quotes around the content types
        if content_type and content_type not in ["\"stripped\"", "\"ip\"", "\"url\""]:
            self.error(model, field, "Content type %s is not allowed" % content_type)

        self.check_modifier_consistent(model, field)
        self.allow_options(model, field,
                           ["blank", "choices", "content_type", "db_index", "default",
                            "max_length", "modifier", "null", "varchar"])

    def validate_field_bool(self, model, field):
        self.check_modifier_consistent(model, field)
        self.allow_options(model, field, ["db_index", "default=True", "default=False", "modifier", "null=False"])
        self.require_options(model, field, ["default"])

    def validate_field_float(self, model, field):
        self.check_modifier_consistent(model, field)
        self.allow_options(model, field, ["blank", "db_index", "default", "modifier", "null"])

    def validate_field_link_onetomany(self, model, field):
        self.check_modifier_consistent(model, field)
        self.allow_options(model, field,
                           ["blank", "db_index", "default", "model", "link_type=manytoone",
                            "modifier", "null", "port", "type=link"])

    def validate_field_link_manytomany(self, model, field):
        self.check_modifier_consistent(model, field)
        self.allow_options(model, field,
                           ["blank", "db_index", "default", "model", "link_type=manytomany",
                            "modifier", "null", "port", "type=link"])

    def validate_field_link(self, model, field):
        link_type = field.get("options", {}).get("link_type")
        if link_type == "manytoone":
            self.validate_field_link_onetomany(model, field)
        elif link_type == "manytomany":
            self.validate_field_link_manytomany(model, field)
        else:
            self.error(model, field, "Unknown link_type %s" % link_type)

    def validate_field_integer(self, model, field):
        # An integer with an option "type=link" is actually a link
        if field.get("options", {}).get("type") == "link":
            self.validate_field_link(model, field)
            return

        self.check_modifier_consistent(model, field)
        self.allow_options(model, field,
                           ["blank", "db_index", "default", "max_value", "min_value", "modifier", "null"])

        if self.is_option_true(field, "blank") and not self.is_option_true(field, "null"):
            self.error(model, field, "If blank is true then null must also be true")

    def validate_field(self, model, field):
        if field["type"] == "string":
            self.validate_field_string(model, field)
        elif field["type"] in ["int32", "uint32"]:
            self.validate_field_integer(model, field)
        elif field["type"] == "float":
            self.validate_field_float(model, field)
        elif field["type"] == "bool":
            self.validate_field_bool(model, field)
        else:
            self.error(model, field, "Unknown field type %s" % field["type"])

    def validate_model(self, model):
        for field in model["fields"]:
            self.validate_field(model, field)

    def validate(self):
        """ Validate all models. This is the main entrypoint for validating xproto. """
        for (name, model) in self.models.items():
            self.validate_model(model)