1327 lines
48 KiB
Python
1327 lines
48 KiB
Python
###
|
|
# Copyright 2020 Hewlett Packard Enterprise, Inc. All rights reserved.
|
|
#
|
|
# 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.
|
|
###
|
|
|
|
# -*- coding: utf-8 -*-
|
|
"""Handles schema and registry gathering as well as schema parsing and validation."""
|
|
|
|
# ---------Imports---------
|
|
|
|
import json
|
|
import logging
|
|
import re
|
|
import textwrap
|
|
|
|
import six
|
|
|
|
from redfish.rest.containers import RisObject
|
|
|
|
from .sharedtypes import JSONEncoder
|
|
|
|
# ---------End of imports---------
|
|
|
|
|
|
# ---------Debug logger---------
|
|
|
|
LOGGER = logging.getLogger(__name__)
|
|
|
|
|
|
# ---------End of debug logger---------
|
|
|
|
|
|
class InvalidPathsError(Exception):
|
|
"""Raised when requested path is not found"""
|
|
|
|
pass
|
|
|
|
|
|
class RegistryValidationError(Exception):
|
|
"""Registration Validation Class Error"""
|
|
|
|
def __init__(self, msg, regentry=None, selector=None):
|
|
super(RegistryValidationError, self).__init__(msg)
|
|
self.reg = regentry
|
|
self.sel = selector
|
|
self.message = msg
|
|
|
|
|
|
class UnknownValidatorError(Exception):
|
|
"""Raised when we find an attribute type that we don't know how to validate."""
|
|
|
|
pass
|
|
|
|
|
|
class ValidationManager(object):
|
|
"""Keeps track of all the schemas and registries and provides helpers
|
|
to simplify validation."""
|
|
|
|
def __init__(self, monolith, defines=None):
|
|
"""init of ValidationManager"""
|
|
super(ValidationManager, self).__init__()
|
|
|
|
self._schemaid = Typepathforval.typepath.schemapath
|
|
self._regid = Typepathforval.typepath.regpath
|
|
|
|
self._classes = list()
|
|
self._classpaths = list()
|
|
# type and path defines object
|
|
self.defines = defines
|
|
self.monolith = monolith
|
|
# error
|
|
self._errors = list()
|
|
self._warnings = list()
|
|
self.updatevalidationdata()
|
|
|
|
@property
|
|
def errors(self):
|
|
"""All errors found by the last validation."""
|
|
return self._errors
|
|
|
|
@property
|
|
def warnings(self):
|
|
"""All warnings found by the last validation."""
|
|
return self._warnings
|
|
|
|
def reset_errors_warnings(self):
|
|
"""Resets warnings and errors, getting ready for the next validation."""
|
|
self._errors = list()
|
|
self._warnings = list()
|
|
|
|
def updatevalidationdata(self):
|
|
"""Loads the types into the validation manager from monolith."""
|
|
monolith = self.monolith
|
|
for instance in monolith.iter():
|
|
if (
|
|
(
|
|
x.lower() in instance.maj_type.lower()
|
|
for x in (
|
|
self.defines.defs.schemafilecollectiontype,
|
|
"Collection.",
|
|
self.defines.defs.regfilecollectiontype,
|
|
)
|
|
)
|
|
and any(x.lower() in instance.path.lower() for x in (self._schemaid, self._regid))
|
|
and instance
|
|
and instance.path not in self._classpaths
|
|
):
|
|
self._classpaths.append(instance.path)
|
|
self._classes.append(instance.resp.dict)
|
|
|
|
def find_prop(self, propname, latestschema=False, proppath=None):
|
|
"""Searches through all locations and returns the schema
|
|
found for the provided propname type.
|
|
|
|
:param propname: String containing the schema name.
|
|
:type propname: str
|
|
:param proppath: String containing the schema path if you wish to use that instead.
|
|
:type proppath: str
|
|
:param latestschema: Flag to determine if we should drop the schema version when we try to
|
|
match schema information. If True, the version will be dropped.
|
|
:type latestschema: bool
|
|
"""
|
|
if proppath:
|
|
self.monolith.load(path=proppath, crawl=False, loadtype="ref")
|
|
return True
|
|
for cls in self._classes:
|
|
found = self.find_property(propname, cls=cls, latestschema=latestschema)
|
|
if found:
|
|
return found
|
|
return None
|
|
|
|
def itermems(self, membername=None):
|
|
"""Searches through all locations and yields each entry.
|
|
|
|
:param membername: string containing the registry name. If not passed we use the typedefines
|
|
string by default.
|
|
:type membername: str
|
|
"""
|
|
if not membername:
|
|
membername = self.defines.defs.collectionstring
|
|
for items in self._classes:
|
|
for item in items[membername]:
|
|
yield item
|
|
|
|
def iterregmems(self, membername=None):
|
|
"""Searches through all registries and yields each entry.
|
|
|
|
:param membername: string containing the registry name. If not passed we use the typedefines
|
|
string by default.
|
|
:type membername: str
|
|
"""
|
|
if not membername:
|
|
membername = self.defines.defs.collectionstring
|
|
for items in self._classes:
|
|
if "registr" in items["Name"].lower(): # For Gen9 type/name issue
|
|
for item in items[membername]:
|
|
yield item
|
|
|
|
def iterschemamems(self, membername=None):
|
|
"""Searches through all schemas and yields each entry
|
|
|
|
:param membername: string containing the registry name. If not passed we use the typedefines
|
|
string by default.
|
|
:type membername: str
|
|
"""
|
|
if not membername:
|
|
membername = self.defines.defs.collectionstring
|
|
for items in self._classes:
|
|
if "schema" in items["Name"].lower(): # For Gen9 type/name issue
|
|
for item in items[membername]:
|
|
yield item
|
|
|
|
def find_property(self, propname, cls=None, latestschema=False):
|
|
"""Returns iLO/BIOS registries/schemas
|
|
|
|
:param propname: string containing the registry/schema name.
|
|
:type propname: str
|
|
:param cls: self._classes list of dictionaries.
|
|
:type cls: list
|
|
:param latestschema: flag to drop the versioning in the type string.
|
|
:type latestschema: bool.
|
|
|
|
:returns: iLO/BIOS registries/schemas that match the supplied name.
|
|
"""
|
|
result = []
|
|
dataloc = cls.get("Items", None)
|
|
dataloc = cls.get("Members", None) if not dataloc else dataloc
|
|
keyword = "Schema"
|
|
if dataloc and isinstance(dataloc, list):
|
|
_ = propname.split(".")[0].strip("#")
|
|
propname = propname.split(".")[0].strip("#") if latestschema else propname
|
|
for entry in dataloc:
|
|
if entry:
|
|
if "Schema" in entry:
|
|
if propname.lower() in entry["Schema"].lower():
|
|
result.append(entry)
|
|
elif "Registry" in entry:
|
|
if (
|
|
propname.lower() in entry["@odata.id"].lower()
|
|
and propname.lower() in entry["Registry"].lower()
|
|
):
|
|
result.append(entry)
|
|
keyword = "Registry"
|
|
else:
|
|
if "@odata.id" in entry:
|
|
reglink = entry["@odata.id"].split("/")
|
|
reglink = reglink[len(reglink) - 2]
|
|
if reglink.lower().startswith(propname.lower()):
|
|
self.monolith.load(path=entry["@odata.id"], crawl=False)
|
|
result.append(self.monolith.paths[entry["@odata.id"]].dict)
|
|
|
|
if result:
|
|
result = max(
|
|
result,
|
|
key=lambda res: (
|
|
res[Typepathforval.typepath.defs.hrefstring]
|
|
if res.get(Typepathforval.typepath.defs.hrefstring, None)
|
|
else res[keyword]
|
|
),
|
|
)
|
|
schemapath = self.geturidict(result["Location"][0])
|
|
self.monolith.load(path=schemapath, crawl=False, loadtype="ref")
|
|
return result
|
|
|
|
def geturidict(self, locationobj):
|
|
"""Return the external reference link.
|
|
|
|
:param locationobj: Dictionary to get the URI reference from.
|
|
:type locationobj: dict
|
|
"""
|
|
if Typepathforval.typepath.defs.isgen10:
|
|
try:
|
|
return locationobj["Uri"]
|
|
except KeyError:
|
|
raise InvalidPathsError("Error accessing Uri path!/n")
|
|
elif Typepathforval.typepath.defs.isgen9:
|
|
try:
|
|
return locationobj["Uri"]["extref"]
|
|
except KeyError:
|
|
raise InvalidPathsError("Error accessing extref path!/n")
|
|
|
|
def validatedict(
|
|
self,
|
|
tdict,
|
|
currtype=None,
|
|
proppath=None,
|
|
latestschema=False,
|
|
searchtype=None,
|
|
monolith=None,
|
|
reg=None,
|
|
unique=None,
|
|
):
|
|
"""Load the schema file and validate tdict against it.
|
|
|
|
:param tdict: the dictionary to test against.
|
|
:type tdict: dict
|
|
:param currtype: String containing the type the tdict dictionary is.
|
|
:type currtype: str
|
|
:param proppath: String containing the schema path of the tdict dictionary if you wish to
|
|
use that instead.
|
|
:type proppath: str
|
|
:param latestschema: Flag to determine if we should drop the schema version when we try to
|
|
match schema information. If True, the version will be dropped.
|
|
:type latestschema: bool
|
|
:param searchtype: Include the attribute registry of you are validating a bios registry.
|
|
:type searchtype: str
|
|
:param monolith: Full data model retrieved from server.
|
|
:type monolith: dict
|
|
:param unique: Flag to override for skipping unique properties.
|
|
:type unique: bool
|
|
:param reg: Registry entry of the given attribute. If this is not provided we will attempt
|
|
to search based on the searchtype and currtype/proppath arguments.
|
|
:type reg: dict.
|
|
:returns: returns an error list.
|
|
"""
|
|
if not reg:
|
|
reg = self.get_registry_model(
|
|
currtype=currtype,
|
|
searchtype=searchtype,
|
|
proppath=proppath,
|
|
latestschema=latestschema,
|
|
)
|
|
|
|
if reg:
|
|
list(
|
|
map(
|
|
lambda x: self.checkreadunique(
|
|
tdict,
|
|
x,
|
|
reg=reg,
|
|
warnings=self._warnings,
|
|
unique=unique,
|
|
searchtype=searchtype,
|
|
),
|
|
list(tdict.keys()),
|
|
)
|
|
)
|
|
orireg = reg.copy()
|
|
ttdict = {key: val for key, val in list(tdict.items()) if not isinstance(val, (dict, list))}
|
|
results = reg.validate_attribute_values(ttdict)
|
|
self._errors.extend(results)
|
|
|
|
for ki, val in list(tdict.items()):
|
|
if ki in ttdict:
|
|
tdict[ki] = ttdict[ki]
|
|
continue
|
|
reg = orireg.copy()
|
|
valexists = False
|
|
if val and isinstance(val, list):
|
|
valexists = True
|
|
# TODO: only validates if its a single dict within list
|
|
if len(val) == 1 and isinstance(val[0], dict):
|
|
treg = self.nestedreg(reg=reg, args=[ki])
|
|
self.validatedict(
|
|
val[0],
|
|
unique=unique,
|
|
monolith=monolith,
|
|
reg=treg,
|
|
currtype=currtype,
|
|
searchtype=searchtype,
|
|
)
|
|
else:
|
|
continue
|
|
elif val and isinstance(val, dict):
|
|
valexists = True
|
|
treg = self.nestedreg(reg=reg, args=[ki])
|
|
self.validatedict(
|
|
val,
|
|
monolith=monolith,
|
|
reg=treg,
|
|
unique=unique,
|
|
searchtype=searchtype,
|
|
)
|
|
if not val and valexists:
|
|
del tdict[ki]
|
|
|
|
else:
|
|
self._errors.append(RegistryValidationError("Unable to locate registry model"))
|
|
return self._errors, self._warnings
|
|
|
|
def checkreadunique(self, tdict, tkey, reg=None, warnings=None, unique=None, searchtype=None):
|
|
"""Check for and remove the readonly and unique attributes if required.
|
|
|
|
:param tdict: the dictionary to test against.
|
|
:type tdict: dict.
|
|
:param tkey: The attribute key value to be tested.
|
|
:type tkey: str.
|
|
:param warnings: list containing found warnings.
|
|
:type warnings: list.
|
|
:param unique: flag to determine override for unique properties.
|
|
:type unique: str.
|
|
:param reg: Registry entry of the given attribute.
|
|
:type reg: dict.
|
|
:returns: returns boolean.
|
|
"""
|
|
if "Attributes" in tdict:
|
|
return False
|
|
# key = list(tdict[tkey])[0]
|
|
# reg = reg["Attributes"][key]
|
|
else:
|
|
if tkey in reg:
|
|
reg = reg[tkey]
|
|
|
|
if not unique and reg.get("IsSystemUniqueProperty", None):
|
|
if tdict[tkey] and not isinstance(tdict[tkey], bool):
|
|
self._warnings.append("Property '%s' is unique and override not authorized. Skipping...\n" % str(tkey))
|
|
del tdict[tkey]
|
|
return True
|
|
if not reg.get("ReadOnly") or (reg.get(tkey, None) and not reg[tkey].get("readonly")):
|
|
if unique and reg.get("IsSystemUniqueProperty", None):
|
|
self._warnings.append("Property '%s' is unique, but override authorized...Patching..\n" % str(tkey))
|
|
return False
|
|
# if not searchtype or (reg.get("ReadOnly") or (reg.get(tkey)
|
|
# and reg[tkey].get("readonly"))):
|
|
# self._warnings.append("Property is read-only. skipping... '%s'" % str(tkey))
|
|
# del tdict[tkey]
|
|
# return True
|
|
elif reg.get("ReadOnly") or (reg.get(tkey) and reg[tkey].get("readonly")):
|
|
if tdict[tkey]:
|
|
self._warnings.append("Property '%s' is read-only. Skipping...\n" % str(tkey))
|
|
del tdict[tkey]
|
|
return True
|
|
else:
|
|
return False
|
|
|
|
def get_registry_model(
|
|
self,
|
|
currtype=None,
|
|
proppath=None,
|
|
getmsg=False,
|
|
searchtype=None,
|
|
newarg=None,
|
|
latestschema=False,
|
|
):
|
|
"""Loads the schema file and find the registry model if available. A registry model is a
|
|
object built for schema/bios registry data.
|
|
|
|
:param currtype: Type selection string.
|
|
:type currtype: dict.
|
|
:param proppath: String containing the schema path if you wish to use that instead.
|
|
:type proppath: str
|
|
:param getmsg: Flag to determine if commit should be skipped.
|
|
:type getmsg: bool
|
|
:param searchtype: Include the attribute registry of you are validating a bios registry.
|
|
:type searchtype: str
|
|
:param newarg: List of multi level properties to be modified.
|
|
:type newarg: list
|
|
:param latestschema: Flag to determine if we should drop the schema version when we try to
|
|
match schema information. If True, the version will be dropped.
|
|
:type latestschema: bool
|
|
:returns: Schema in object form called a registry object.
|
|
"""
|
|
regdict = None
|
|
monolith = self.monolith
|
|
currtype = currtype.split("#")[-1].split(".")[0] + "." if currtype and latestschema else currtype
|
|
if (
|
|
not currtype
|
|
or not self.find_prop(
|
|
currtype,
|
|
latestschema=latestschema,
|
|
proppath=proppath if not searchtype else None,
|
|
)
|
|
) and (not searchtype):
|
|
self._errors.append(RegistryValidationError("Location info is missing.\n"))
|
|
return None
|
|
if not searchtype:
|
|
searchtype = "object"
|
|
|
|
try:
|
|
for instance in monolith.iter(searchtype):
|
|
if (
|
|
(searchtype == Typepathforval.typepath.defs.attributeregtype)
|
|
or (
|
|
searchtype == "object"
|
|
and any(
|
|
currtype in xtitle
|
|
for xtitle in (
|
|
instance.resp.dict.get("title", ""),
|
|
instance.resp.dict.get("oldtitle", ""),
|
|
)
|
|
)
|
|
)
|
|
or (
|
|
searchtype != "object"
|
|
and currtype.split("#")[-1].split(".")[0] == instance.dict.get("RegistryPrefix", "")
|
|
)
|
|
):
|
|
regdict = instance.resp.dict
|
|
break
|
|
except BaseException:
|
|
pass
|
|
|
|
if not regdict:
|
|
self._errors.append(RegistryValidationError("Location data is empty.\n"))
|
|
return None
|
|
|
|
jsonreg = json.loads(json.dumps(regdict, indent=2, cls=JSONEncoder))
|
|
|
|
if getmsg:
|
|
return {jsonreg["RegistryPrefix"]: jsonreg["Messages"]}
|
|
|
|
# This was done for bios registry model compatibility
|
|
if "RegistryEntries" in jsonreg:
|
|
regitem = jsonreg["RegistryEntries"]
|
|
if "Attributes" in regitem:
|
|
newitem = {item[Typepathforval.typepath.defs.attributenametype]: item for item in regitem["Attributes"]}
|
|
regitem["Attributes"] = newitem
|
|
if not Typepathforval.typepath.flagiften:
|
|
del regitem["Attributes"]
|
|
newitem.update(regitem)
|
|
regitem = newitem
|
|
reg = HpPropertiesRegistry.parse(regitem)
|
|
return self.nestedreg(reg=reg, args=newarg) if newarg else reg
|
|
|
|
if "properties" in jsonreg:
|
|
regitem = jsonreg["properties"]
|
|
if "Properties" in regitem:
|
|
regitem.update(regitem["Properties"])
|
|
del regitem["Properties"]
|
|
reg = HpPropertiesRegistry.parse(regitem)
|
|
|
|
return self.nestedreg(reg=reg, args=newarg) if newarg else reg
|
|
|
|
def nestedreg(self, reg=None, args=None):
|
|
"""Go through the registry entry to find the required nested attribute.
|
|
|
|
:param reg: Registry entry of the given attribute.
|
|
:type reg: dict
|
|
:param args: List of multi level properties to be modified.
|
|
:type args: list
|
|
:returns: dict of Registry entry
|
|
"""
|
|
for arg in args:
|
|
try:
|
|
arg = next((key for key in list(reg.keys()) if key.lower() == arg.lower()), None)
|
|
if not arg:
|
|
return None
|
|
if ("properties" in reg[arg].keys()) and ("patternProperties" in reg[arg].keys()):
|
|
reg[arg]["properties"].update(reg[arg]["patternProperties"])
|
|
reg = reg[arg]["properties"]
|
|
elif "oneOf" in reg[arg]:
|
|
oneof = reg[arg]["oneOf"]
|
|
for item in oneof:
|
|
reg = item["properties"]
|
|
elif (
|
|
"type" in reg[arg]
|
|
and reg[arg]["type"] == "array"
|
|
and "items" in reg[arg]
|
|
and "properties" in reg[arg]["items"]
|
|
):
|
|
reg = reg[arg]["items"]["properties"]
|
|
elif ("properties" not in reg[arg].keys()) or ("patternProperties" in reg[arg].keys()):
|
|
reg = reg[arg]
|
|
else:
|
|
reg = reg[arg]["properties"]
|
|
except:
|
|
try:
|
|
reg = reg[arg]["patternProperties"]
|
|
except:
|
|
return None
|
|
return reg
|
|
|
|
|
|
class HpPropertiesRegistry(RisObject):
|
|
"""Models a schema or bios attribute registry. Registry model."""
|
|
|
|
def __init__(self, d):
|
|
super(HpPropertiesRegistry, self).__init__(d)
|
|
|
|
def validate_attribute_values(self, tdict):
|
|
"""Look for tdict in the attribute list and attempt to validate its value.
|
|
|
|
:param tdict: the dictionary to test against.
|
|
:type tdict: dict
|
|
:returns: A validated list
|
|
"""
|
|
result = list()
|
|
|
|
for tkey in tdict:
|
|
if tkey not in self:
|
|
# Added for Gen 9 Bios properties not in registry
|
|
continue
|
|
elif self[tkey] and (checkattr(self[tkey], "type") or checkattr(self[tkey], "Type")):
|
|
keyval = list()
|
|
keyval.append(tdict[tkey])
|
|
temp = self.validate_attribute(self[tkey], keyval, tkey)
|
|
tdict[tkey] = keyval[0]
|
|
|
|
for err in temp:
|
|
if isinstance(err, RegistryValidationError):
|
|
if err.reg:
|
|
err.sel = tkey
|
|
|
|
result.extend(temp)
|
|
|
|
return result
|
|
|
|
def get_validator(self, attrname, newargs=None, oneof=None):
|
|
"""Returns attribute validator type.
|
|
|
|
:param attrname: attribute name to validate. Ex: In A/B/C, this will be A.
|
|
:type attrname: str
|
|
:param newargs: List of multi level properties to be modified. Ex: In A/B/C this will be
|
|
a list of B and C.
|
|
:type newargs: list
|
|
:param oneof: Special string for "oneof" options within validation.
|
|
:type oneof: string
|
|
:returns: The validator type class for the property passed.
|
|
"""
|
|
if oneof:
|
|
self = oneof
|
|
|
|
if newargs:
|
|
for arg in newargs:
|
|
try:
|
|
self = self["properties"]
|
|
except Exception:
|
|
pass
|
|
|
|
if not checkattr(self, arg):
|
|
return None
|
|
elif not arg == newargs[-1]:
|
|
self = self[arg]
|
|
|
|
if not checkattr(self, attrname):
|
|
return None
|
|
|
|
validator = None
|
|
if EnumValidator.is_type(self[attrname]):
|
|
validator = EnumValidator.parse(self[attrname])
|
|
elif StringValidator.is_type(self[attrname]):
|
|
validator = StringValidator.parse(self[attrname])
|
|
elif ObjectValidator.is_type(self[attrname]):
|
|
validator = ObjectValidator.parse(self[attrname])
|
|
elif IntegerValidator.is_type(self[attrname]):
|
|
validator = IntegerValidator.parse(self[attrname])
|
|
elif BoolValidator.is_type(self[attrname]):
|
|
validator = BoolValidator.parse(self[attrname])
|
|
elif PasswordValidator.is_type(self[attrname]):
|
|
validator = PasswordValidator.parse(self[attrname])
|
|
elif "oneOf" in list(self[attrname].keys()):
|
|
for item in self[attrname]["oneOf"]:
|
|
validator = self.get_validator(attrname, newargs, HpPropertiesRegistry({attrname: item}))
|
|
if validator:
|
|
break
|
|
return validator
|
|
|
|
def validate_attribute(self, attrentry, attrvallist, name):
|
|
"""Function to validate attribute against its schema.
|
|
|
|
:param attrentry: Key of property to validate.
|
|
:type attrentry: str
|
|
:param attrval: Value of Key to validate.
|
|
:type attrval: str
|
|
:param name: Clean name for outputting information to users.
|
|
:type name: str
|
|
:returns: returns list with validated attributes
|
|
"""
|
|
result = list()
|
|
validator = None
|
|
if self.nulltypevalidationcheck(attrval=attrvallist[0], attrentry=attrentry):
|
|
return result
|
|
|
|
if EnumValidator.is_type(attrentry):
|
|
validator = EnumValidator.parse(attrentry)
|
|
elif StringValidator.is_type(attrentry):
|
|
validator = StringValidator.parse(attrentry)
|
|
elif IntegerValidator.is_type(attrentry):
|
|
validator = IntegerValidator.parse(attrentry)
|
|
elif BoolValidator.is_type(attrentry):
|
|
validator = BoolValidator.parse(attrentry)
|
|
elif ObjectValidator.is_type(attrentry):
|
|
validator = ObjectValidator.parse(attrentry)
|
|
elif PasswordValidator.is_type(attrentry):
|
|
validator = PasswordValidator.parse(attrentry)
|
|
else:
|
|
raise UnknownValidatorError(attrentry)
|
|
|
|
if validator:
|
|
result.extend(validator.is_array(attrentry, attrvallist, name))
|
|
result.extend(validator.validate(attrvallist, name))
|
|
return result
|
|
|
|
def nulltypevalidationcheck(self, attrval=None, attrentry=None):
|
|
"""Function to validate null attributes against iLO schema
|
|
|
|
:param attrentry: Key of property to validate.
|
|
:type attrentry: str
|
|
:param attrval: Value of Key to validate.
|
|
:type attrval: str
|
|
:returns: True if entry is null and valid.
|
|
"""
|
|
if "type" in attrentry and attrval is None:
|
|
if isinstance(attrentry["type"], list):
|
|
for item in attrentry["type"]:
|
|
if item.lower() == "null":
|
|
return True
|
|
return False
|
|
|
|
|
|
class BaseValidator(RisObject):
|
|
"""Base class for all validators."""
|
|
|
|
def __init__(self, d):
|
|
super(BaseValidator, self).__init__(d)
|
|
|
|
def validate(self):
|
|
"""Overridable function for validation"""
|
|
raise RuntimeError("You must override this method in your derived class")
|
|
|
|
def common_print_help(self, name):
|
|
"""Common human readable schema data.
|
|
|
|
:param name: clean name for outputting.
|
|
:type name: str
|
|
:returns: A human readable string of schema data.
|
|
"""
|
|
outdata = ""
|
|
wrapper = textwrap.TextWrapper()
|
|
wrapper.initial_indent = " " * 4
|
|
wrapper.subsequent_indent = " " * 4
|
|
|
|
outdata += "\nNAME\n"
|
|
outdata += "%s\n" % wrapper.fill("%s" % name)
|
|
outdata += "\n"
|
|
|
|
if "DisplayName" in self:
|
|
outdata += "\nDISPLAY NAME\n"
|
|
outdata += "%s\n" % wrapper.fill("%(DisplayName)s" % self)
|
|
outdata += "\n"
|
|
|
|
if "description" in self:
|
|
outdata += "\nDESCRIPTION\n"
|
|
outdata += "%s\n" % wrapper.fill("%(description)s" % self)
|
|
outdata += "\n"
|
|
|
|
if "HelpText" in self:
|
|
outdata += "\nHELP TEXT\n"
|
|
outdata += "%s\n" % wrapper.fill("%(HelpText)s" % self)
|
|
outdata += "\n"
|
|
|
|
if "WarningText" in self:
|
|
outdata += "\n************************************************\n"
|
|
outdata += "\nWARNING\n"
|
|
outdata += "%s\n" % wrapper.fill("%(WarningText)s" % self)
|
|
outdata += "\n\n**********************************************\n"
|
|
outdata += "\n"
|
|
|
|
if "type" in self and isinstance(self["type"], list):
|
|
outdata += "\nTYPE\n"
|
|
for item in self["type"]:
|
|
outdata += "%s\n" % wrapper.fill("%s" % item)
|
|
outdata += "\n"
|
|
elif "type" in self:
|
|
outdata += "\nTYPE\n"
|
|
outdata += "%s\n" % wrapper.fill("%(type)s" % self)
|
|
outdata += "\n"
|
|
elif "Type" in self:
|
|
outdata += "\nTYPE\n"
|
|
outdata += "%s\n" % wrapper.fill("%(Type)s" % self)
|
|
outdata += "\n"
|
|
|
|
if "ReadOnly" in self:
|
|
outdata += "\nREAD-ONLY\n"
|
|
outdata += "%s\n" % wrapper.fill("%(ReadOnly)s" % self)
|
|
outdata += "\n"
|
|
elif "readonly" in self:
|
|
outdata += "\nREAD-ONLY\n"
|
|
outdata += "%s\n" % wrapper.fill("%(readonly)s" % self)
|
|
outdata += "\n"
|
|
return outdata
|
|
|
|
def is_arrtype(self, attrentry):
|
|
"""Validate that the type is an array.
|
|
|
|
:param attrentry: Registry model entry used for validation.
|
|
:type attrentry: dict
|
|
:returns: A boolean based on whether type is an array.
|
|
"""
|
|
if "type" in attrentry and attrentry["type"] == "array":
|
|
return True
|
|
return False
|
|
|
|
def is_array(self, attrentry, arrval, name):
|
|
"""Validate that the given value is an array type.
|
|
|
|
:param attrentry: Registry model entry used for validation.
|
|
:type attrentry: dict
|
|
:param attrval: Value of Key to validate.
|
|
:type attrval: str
|
|
:returns: A boolean based on whether type is array and the value is valid for array type.
|
|
"""
|
|
result = []
|
|
|
|
if self.is_arrtype(attrentry):
|
|
if isinstance(
|
|
arrval[0],
|
|
(
|
|
frozenset,
|
|
list,
|
|
set,
|
|
tuple,
|
|
),
|
|
):
|
|
return []
|
|
else:
|
|
result.append(
|
|
RegistryValidationError(
|
|
"'%s' is not a valid setting " "for '%s', expecting an array" % (arrval[0], name),
|
|
regentry=self,
|
|
)
|
|
)
|
|
return result
|
|
|
|
|
|
class EnumValidator(BaseValidator):
|
|
"""Enum validator class"""
|
|
|
|
def __init__(self, d):
|
|
super(EnumValidator, self).__init__(d)
|
|
|
|
@staticmethod
|
|
def is_type(attrentry):
|
|
"""Validate that the type is enumeration.
|
|
|
|
:param attrentry: Registry model entry used for validation.
|
|
:type attrentry: dict
|
|
:returns: A boolean based on whether type is eneumeration.
|
|
"""
|
|
if "type" in attrentry:
|
|
if isinstance(attrentry["type"], list):
|
|
for item in attrentry["type"]:
|
|
if item.lower() == "enumeration":
|
|
return True
|
|
elif "enum" in attrentry and item.lower() == "string":
|
|
return True
|
|
elif "enum" in attrentry and attrentry["type"] == "array":
|
|
for key, value in attrentry["items"].items():
|
|
if key.lower() == "type" and value.lower() == "string":
|
|
return True
|
|
else:
|
|
if attrentry["type"].lower() == "enumeration":
|
|
return True
|
|
elif "enum" in attrentry and attrentry["type"].lower() == "string":
|
|
return True
|
|
elif "Type" in attrentry:
|
|
if attrentry["Type"].lower() == "enumeration":
|
|
return True
|
|
|
|
return False
|
|
|
|
def validate(self, keyval, name):
|
|
"""Validate against schemas.
|
|
|
|
:param keyval: New value to be used for validation in a list
|
|
:type keyval: list
|
|
:param name: Clean name for outputting human readable info.
|
|
:type name: str
|
|
:returns: An error if validation fails.
|
|
"""
|
|
result = list()
|
|
newval = keyval[0]
|
|
|
|
try:
|
|
for possibleval in self.enum:
|
|
if (
|
|
possibleval
|
|
and (
|
|
isinstance(possibleval, type(newval))
|
|
or (isinstance(possibleval, six.string_types) and isinstance(newval, six.string_types))
|
|
)
|
|
and possibleval.lower() == str(newval).lower()
|
|
):
|
|
keyval[0] = possibleval
|
|
return result
|
|
except Exception:
|
|
for possibleval in self.Value:
|
|
if possibleval.ValueName.lower() == str(newval).lower():
|
|
keyval[0] = possibleval.ValueName
|
|
return result
|
|
|
|
result.append(
|
|
RegistryValidationError("'%s' is not a valid setting " "for '%s'" % (newval, name), regentry=self)
|
|
)
|
|
|
|
return result
|
|
|
|
def print_help(self, name):
|
|
"""Human readable schema information specific to Enum data.
|
|
|
|
:param name: Clean name for outputting human readable info.
|
|
:type name: str
|
|
:returns: A human readable string of schema data.
|
|
"""
|
|
outdata = self.common_print_help(name)
|
|
outdata += "\nPOSSIBLE VALUES\n"
|
|
try:
|
|
for possibleval in self.enum:
|
|
outdata += " %s\n" % possibleval
|
|
except Exception:
|
|
for possibleval in self.Value:
|
|
outdata += " %(ValueName)s\n" % possibleval
|
|
outdata += "\n"
|
|
return outdata
|
|
|
|
|
|
class BoolValidator(BaseValidator):
|
|
"""Bool validator class"""
|
|
|
|
def __init__(self, d):
|
|
super(BoolValidator, self).__init__(d)
|
|
|
|
@staticmethod
|
|
def is_type(attrentry):
|
|
"""Validate that the type is boolean.
|
|
|
|
:param attrentry: Registry model entry used for validation.
|
|
:type attrentry: dict
|
|
:returns: A boolean based on whether type is boolean.
|
|
"""
|
|
if "type" in attrentry:
|
|
if isinstance(attrentry["type"], list):
|
|
for item in attrentry["type"]:
|
|
if item.lower() == "boolean":
|
|
return True
|
|
elif attrentry["type"] == "array":
|
|
for key, value in attrentry["items"].items():
|
|
if key.lower() == "type" and value.lower() == "boolean":
|
|
return True
|
|
else:
|
|
if attrentry["type"].lower() == "boolean":
|
|
return True
|
|
elif "Type" in attrentry:
|
|
if attrentry["Type"].lower() == "boolean":
|
|
return True
|
|
|
|
return False
|
|
|
|
def validate(self, newval, name):
|
|
"""Validate against schemas.
|
|
|
|
:param newval: New value to be used for validation in a list
|
|
:type newval: list
|
|
:param name: Clean name for outputting human readable info.
|
|
:type name: str
|
|
:returns: An error if value is invalid.
|
|
"""
|
|
result = list()
|
|
if newval[0] is False or newval[0] is True:
|
|
return result
|
|
|
|
result.append(
|
|
RegistryValidationError("'%s' is not a valid setting for '%s'" % (newval[0], name), regentry=self)
|
|
)
|
|
|
|
return result
|
|
|
|
def print_help(self, name):
|
|
"""Human readable schema information specific to Boolean data.
|
|
|
|
:param name: Clean name for outputting human readable info.
|
|
:type name: str
|
|
:returns: A human readable string of schema data.
|
|
"""
|
|
outdata = self.common_print_help(name)
|
|
outdata += "\nPOSSIBLE VALUES\n"
|
|
outdata += " True or False\n"
|
|
outdata += "\n"
|
|
return outdata
|
|
|
|
|
|
class StringValidator(BaseValidator):
|
|
"""Constructor"""
|
|
|
|
def __init__(self, d):
|
|
super(StringValidator, self).__init__(d)
|
|
|
|
@staticmethod
|
|
def is_type(attrentry):
|
|
"""Validate that the type is string.
|
|
|
|
:param attrentry: Registry model entry used for validation.
|
|
:type attrentry: dict
|
|
:returns: A boolean based on whether type is string.
|
|
"""
|
|
if "type" in attrentry:
|
|
if isinstance(attrentry["type"], list):
|
|
for item in attrentry["type"]:
|
|
if item.lower() == "string":
|
|
return True
|
|
elif attrentry["type"] == "array":
|
|
for key, value in attrentry["items"].items():
|
|
if key.lower() == "type" and "string" in value:
|
|
return True
|
|
else:
|
|
if attrentry["type"].lower() == "string":
|
|
return True
|
|
elif "Type" in attrentry:
|
|
if attrentry["Type"].lower() == "string":
|
|
return True
|
|
|
|
return False
|
|
|
|
def validate(self, newvallist, _):
|
|
"""Validate against schemas.
|
|
|
|
:param newvallist: New value to be used for validation in a list
|
|
:type newvallist: list
|
|
:returns: An error if validation fails criteria.
|
|
"""
|
|
newval = newvallist[0]
|
|
result = list()
|
|
namestr = Typepathforval.typepath.defs.attributenametype
|
|
if not isinstance(newval, str):
|
|
result.append(RegistryValidationError("Given value must be a string"))
|
|
return result
|
|
if "MinLength" in self:
|
|
if len(newval) < int(self["MinLength"]):
|
|
result.append(
|
|
RegistryValidationError(
|
|
"'%s' must be at least '%s' characters long" % (self[namestr], int(self["MinLength"])),
|
|
regentry=self,
|
|
)
|
|
)
|
|
|
|
if "MaxLength" in self:
|
|
if len(newval) > int(self["MaxLength"]):
|
|
result.append(
|
|
RegistryValidationError(
|
|
"'%s' must be less than '%s' characters long" % (self[namestr], int(self["MaxLength"])),
|
|
regentry=self,
|
|
)
|
|
)
|
|
|
|
if "ValueExpression" in self:
|
|
if self["ValueExpression"]:
|
|
pat = re.compile(self["ValueExpression"])
|
|
if newval and not pat.match(newval):
|
|
result.append(
|
|
RegistryValidationError(
|
|
"'%s' must match the regular expression " "'%s'" % (self[namestr], self["ValueExpression"]),
|
|
regentry=self,
|
|
)
|
|
)
|
|
|
|
return result
|
|
|
|
def print_help(self, name):
|
|
"""Human readable schema information specific to String data.
|
|
|
|
:param name: Clean name for outputting human readable info.
|
|
:type name: str
|
|
:returns: A human readable string of schema data.
|
|
"""
|
|
wrapper = textwrap.TextWrapper()
|
|
wrapper.initial_indent = " " * 4
|
|
wrapper.subsequent_indent = " " * 4
|
|
outdata = self.common_print_help(name)
|
|
if "MinLength" in self:
|
|
outdata += "\nMIN LENGTH\n"
|
|
outdata += "%s" % wrapper.fill("%(MinLength)s" % self)
|
|
outdata += "\n"
|
|
|
|
if "MaxLength" in self:
|
|
outdata += "\nMAX LENGTH\n"
|
|
outdata += "%s" % wrapper.fill("%(MaxLength)s" % self)
|
|
outdata += "\n"
|
|
return outdata
|
|
|
|
|
|
class IntegerValidator(BaseValidator):
|
|
"""Interger validator class"""
|
|
|
|
def __init__(self, d):
|
|
super(IntegerValidator, self).__init__(d)
|
|
|
|
@staticmethod
|
|
def is_type(attrentry):
|
|
"""Validate that the type is integer.
|
|
|
|
:param attrentry: Registry model entry used for validation.
|
|
:type attrentry: dict
|
|
:returns: A boolean based on whether type is integer.
|
|
"""
|
|
if "type" in attrentry:
|
|
if isinstance(attrentry["type"], list):
|
|
for item in attrentry["type"]:
|
|
if item.lower() == "integer" or item.lower() == "number":
|
|
return True
|
|
elif attrentry["type"] == "array":
|
|
for key, value in attrentry["items"].items():
|
|
if key.lower() == "type":
|
|
if value.lower() == "integer" or value.lower() == "number":
|
|
return True
|
|
else:
|
|
if attrentry["type"].lower() == "integer" or attrentry["type"].lower().lower() == "number":
|
|
return True
|
|
elif "Type" in attrentry:
|
|
if attrentry["Type"].lower() == "integer":
|
|
return True
|
|
|
|
return False
|
|
|
|
def validate(self, newvallist, _):
|
|
"""Validate against schemas.
|
|
|
|
:param newvallist: New value to be used for validation in a list
|
|
:type newvallist: list
|
|
:returns: An error if validation fails criteria.
|
|
"""
|
|
result = list()
|
|
try:
|
|
intval = int(newvallist[0])
|
|
newvallist[0] = intval
|
|
except:
|
|
result.append(RegistryValidationError("'%(Name)s' must " "be an integer value'" % (self), regentry=self))
|
|
return result
|
|
|
|
if newvallist[0] and not str(intval).isdigit():
|
|
result.append(RegistryValidationError("'%(Name)s' must " "be an integer value'" % (self), regentry=self))
|
|
return result
|
|
|
|
if "LowerBound" in self:
|
|
if intval < int(self["LowerBound"]):
|
|
result.append(
|
|
RegistryValidationError(
|
|
"'%s' must be greater" " than or equal to '%s'" % (self.Name, int(self["LowerBound"])),
|
|
regentry=self,
|
|
)
|
|
)
|
|
|
|
if "UpperBound" in self:
|
|
if intval > int(self["UpperBound"]):
|
|
result.append(
|
|
RegistryValidationError(
|
|
"'%s' must be less " "than or equal to '%s'" % (self.Name, int(self["LowerBound"])),
|
|
regentry=self,
|
|
)
|
|
)
|
|
|
|
return result
|
|
|
|
def print_help(self, name):
|
|
"""Human readable schema information specific to Integer data.
|
|
|
|
:param name: Clean name for outputting human readable info.
|
|
:type name: str
|
|
:returns: A human readable string of schema data.
|
|
"""
|
|
outdata = self.common_print_help(name)
|
|
return outdata
|
|
|
|
|
|
class ObjectValidator(BaseValidator):
|
|
"""Object validator class"""
|
|
|
|
def __init__(self, d):
|
|
super(ObjectValidator, self).__init__(d)
|
|
|
|
@staticmethod
|
|
def is_type(attrentry):
|
|
"""Validate that the type is object.
|
|
|
|
:param attrentry: Registry model entry used for validation.
|
|
:type attrentry: dict
|
|
:returns: A boolean based on whether type is object.
|
|
"""
|
|
if "type" in attrentry:
|
|
if isinstance(attrentry["type"], list):
|
|
for item in attrentry["type"]:
|
|
if item.lower() == "object":
|
|
return True
|
|
elif attrentry["type"] == "array":
|
|
for key, value in attrentry["items"].items():
|
|
if key.lower() == "type" and value.lower() == "object":
|
|
return True
|
|
elif key.lower() == "anyof":
|
|
try:
|
|
if value[0]["type"] == "object":
|
|
return True
|
|
except Exception:
|
|
continue
|
|
else:
|
|
if attrentry["type"].lower() == "object":
|
|
return True
|
|
elif "Type" in attrentry:
|
|
if attrentry["Type"].lower() == "object":
|
|
return True
|
|
|
|
return False
|
|
|
|
def validate(self, newval, name):
|
|
"""Validate against schemas.
|
|
|
|
:param newval: New value to be used for validation in a list
|
|
:type newval: list
|
|
:param name: Clean name for outputting human readable info.
|
|
:type name: str
|
|
:returns: An error if value is invalid.
|
|
"""
|
|
# TODO: need to add logic for true postive and false negatives.
|
|
result = list()
|
|
if isinstance(newval[0], (dict, six.string_types, int)):
|
|
result.append(
|
|
RegistryValidationError(
|
|
"'%s' is not a valid setting for '%s'" % (newval[0], name),
|
|
regentry=self,
|
|
)
|
|
)
|
|
return result
|
|
|
|
def print_help(self, name):
|
|
"""Human readable schema information specific to Object data.
|
|
|
|
:param name: Clean name for outputting human readable info.
|
|
:type name: str
|
|
:returns: A human readable string of schema data.
|
|
"""
|
|
wrapper = textwrap.TextWrapper()
|
|
wrapper.initial_indent = " " * 4
|
|
wrapper.subsequent_indent = " " * 4
|
|
outdata = self.common_print_help(name)
|
|
if "properties" in self:
|
|
outdata += "\nSUB-PROPERTIES\n"
|
|
propdata = ", ".join(self.properties.keys())
|
|
outdata += "%s" % wrapper.fill("%s" % propdata)
|
|
outdata += "\n"
|
|
elif "items" in self:
|
|
outdata += "\nSUB-PROPERTIES\n"
|
|
propdata = ", ".join(self["items"].properties.keys())
|
|
outdata += "%s" % wrapper.fill("%s" % propdata)
|
|
outdata += "\n"
|
|
|
|
return outdata
|
|
|
|
|
|
class PasswordValidator(BaseValidator):
|
|
"""Password validator class"""
|
|
|
|
def __init__(self, d):
|
|
super(PasswordValidator, self).__init__(d)
|
|
|
|
@staticmethod
|
|
def is_type(attrentry):
|
|
"""Validate that the type is a password.
|
|
|
|
:param attrentry: Registry model entry used for validation.
|
|
:type attrentry: dict
|
|
:returns: A boolean based on whether type is a password.
|
|
"""
|
|
if "type" in attrentry:
|
|
if isinstance(attrentry["type"], list):
|
|
for item in attrentry["type"]:
|
|
if item.lower() == "password":
|
|
return True
|
|
elif attrentry["type"] == "array":
|
|
for key, value in attrentry["items"].items():
|
|
if key.lower() == "type" and value.lower() == "password":
|
|
return True
|
|
else:
|
|
if attrentry["type"].lower() == "password":
|
|
return True
|
|
elif "Type" in attrentry:
|
|
if attrentry["Type"].lower() == "password":
|
|
return True
|
|
|
|
return False
|
|
|
|
def validate(self, newvallist, _):
|
|
"""Validate against schemas.
|
|
|
|
:param newvallist: New value to be used for validation in a list
|
|
:type newvallist: list
|
|
:returns: An error if validation fails criteria.
|
|
"""
|
|
result = list()
|
|
newval = newvallist[0]
|
|
|
|
if newval is None:
|
|
return result
|
|
|
|
if not isinstance(newval, str):
|
|
result.append(RegistryValidationError("Given value must be a string"))
|
|
if "MinLength" in self:
|
|
if len(newval) < int(self["MinLength"]):
|
|
result.append(
|
|
RegistryValidationError(
|
|
"'%s' must be at least" " '%s' characters long" % (self.Name, int(self["MinLength"])),
|
|
regentry=self,
|
|
)
|
|
)
|
|
|
|
if "MaxLength" in self:
|
|
if len(newval) > int(self["MaxLength"]):
|
|
result.append(
|
|
RegistryValidationError(
|
|
"'%s' must be less " "than '%s' characters long" % (self.Name, int(self["MaxLength"])),
|
|
regentry=self,
|
|
)
|
|
)
|
|
|
|
if "ValueExpression" in self:
|
|
if self["ValueExpression"]:
|
|
pat = re.compile(self["ValueExpression"])
|
|
if newval and not pat.match(newval):
|
|
result.append(
|
|
RegistryValidationError(
|
|
"'%(Name)s' must " "match the regular expression '%(Value" "Expression)s'" % (self),
|
|
regentry=self,
|
|
)
|
|
)
|
|
|
|
return result
|
|
|
|
def print_help(self, name):
|
|
"""Human readable schema information specific to Password data.
|
|
|
|
:param name: Clean name for outputting human readable info.
|
|
:type name: str
|
|
:returns: A human readable string of schema data.
|
|
"""
|
|
wrapper = textwrap.TextWrapper()
|
|
wrapper.initial_indent = " " * 4
|
|
wrapper.subsequent_indent = " " * 4
|
|
outdata = self.common_print_help(name)
|
|
if "MinLength" in self:
|
|
outdata += "\nMIN LENGTH\n"
|
|
outdata += "%s" % wrapper.fill("%(MinLength)s" % self)
|
|
outdata += "\n"
|
|
|
|
if "MaxLength" in self:
|
|
outdata += "\nMAX LENGTH\n"
|
|
outdata += "%s" % wrapper.fill("%(MaxLength)s" % self)
|
|
outdata += "\n"
|
|
return outdata
|
|
|
|
|
|
class Typepathforval(object):
|
|
"""Way to store the typepath defines object."""
|
|
|
|
typepath = None
|
|
|
|
def __new__(cls, typepathobj):
|
|
if typepathobj:
|
|
Typepathforval.typepath = typepathobj
|
|
|
|
|
|
def checkattr(aobj, prop):
|
|
"""Check if an attribute exists"""
|
|
try:
|
|
if hasattr(aobj, prop):
|
|
return True
|
|
except:
|
|
pass
|
|
return False
|