# # Copyright (c), 2016-2024, SISSA (International School for Advanced Studies). # All rights reserved. # This file is distributed under the terms of the MIT License. # See the file 'LICENSE' in the root directory of the present # distribution, or http://opensource.org/licenses/MIT. # # @author Davide Brunato # import copy import threading import warnings from collections.abc import Collection, Iterator, Iterable from contextlib import contextmanager from functools import cached_property from itertools import dropwhile from typing import Any, cast, Optional, Type from elementpath import XPathToken import xmlschema.names as nm from xmlschema.aliases import SchemaType, BaseXsdType, LocationsType, \ SchemaGlobalType, SchemaSourceType, NsmapType, StagedItemType, ComponentClassType from xmlschema.exceptions import XMLSchemaAttributeError, XMLSchemaTypeError, \ XMLSchemaValueError, XMLSchemaWarning, XMLSchemaNamespaceError, XMLSchemaException from xmlschema.translation import gettext as _ from xmlschema.utils.misc import deprecated from xmlschema.utils.qnames import get_extended_qname from xmlschema.utils.descriptors import validator_property from xmlschema.utils.urls import get_url, normalize_url from xmlschema.namespaces import NamespaceResourcesMap from xmlschema.loaders import SchemaLoader from xmlschema.resources import XMLResource from .exceptions import XMLSchemaModelError, XMLSchemaModelDepthError, XMLSchemaParseError from .xsdbase import XsdValidator, XsdComponent from .models import check_model from . import XsdAttribute, XsdSimpleType, XsdComplexType, XsdElement, \ XsdGroup, XsdIdentity, XsdUnion, XsdAtomicRestriction, \ XsdAtomic, XsdAtomicBuiltin, XsdNotation, XsdAttributeGroup from .builders import GLOBAL_MAP_ATTRIBUTE, GlobalMaps # Default placeholder for deprecation of argument 'validation' in XsdGlobals _strict = type('str', (str,), {})('strict') class XsdGlobals(XsdValidator, Collection[SchemaType]): """ Mediator collection class for composing XML schema instances and provides lookup maps. It stores the global declarations defined in the registered schemas. Register a schema to add its declarations to the global maps. :param validator: the origin schema class/instance used for creating the global maps. :param validation: deprecated argument for the validation mode, now it takes the \ validation mode of the validator. :param parent: an optional parent schema, that is required to be built and with \ no use of the target namespace of the validator. :param loader_class: an optional subclass of :class:`SchemaLoader` to use for creating \ the loader instance. :param locations: schema extra location hints, that can include custom resource locations \ or additional namespaces to import after processing schema's import statements. :param use_fallback: if `True` the schema processor uses the validator fallback \ location hints to load well-known namespaces (e.g. xhtml). :param use_xpath3: if `True` an XSD 1.1 schema instance uses the XPath 3 processor \ for assertions. For default a full XPath 2.0 processor is used. """ _schemas: set[SchemaType] namespaces: NamespaceResourcesMap[SchemaType] loader: SchemaLoader substitution_groups: dict[str, set[XsdElement]] identities: dict[str, XsdIdentity] __slots__ = ('_build_lock', '_schemas', '_parent', 'validation', 'errors', 'validator', 'namespaces', 'loader', 'global_maps', 'types', 'notations', 'attributes', 'attribute_groups', 'elements', 'groups', 'substitution_groups', 'identities', '_built') def __init__(self, validator: SchemaType, validation: str = _strict, parent: Optional[SchemaType] = None, loader_class: Optional[Type[SchemaLoader]] = None, locations: Optional[LocationsType] = None, use_fallback: bool = True, use_xpath3: bool = False) -> None: if not isinstance(validation, _strict.__class__): msg = "argument 'validation' is not used and will be removed in v5.0" warnings.warn(msg, DeprecationWarning, stacklevel=1) super().__init__(validator.validation) self._build_lock = threading.Lock() self._built = False self._schemas = set() self._parent = parent self.validator = validator self.namespaces = NamespaceResourcesMap() # Registered schemas by namespace URI self.global_maps = GlobalMaps.from_builders(validator.builders) (self.types, self.notations, self.attributes, self.attribute_groups, self.elements, self.groups) = self.global_maps self.substitution_groups = {} self.identities = {} for ancestor in self.iter_ancestors(): self._schemas.update(ancestor.maps.schemas) self.namespaces.update(ancestor.maps.namespaces) ancestor.maps.build() self.global_maps.update(ancestor.maps.global_maps) self.substitution_groups.update(ancestor.maps.substitution_groups) self.identities.update(ancestor.maps.identities) self.validator.maps = self self.loader = (loader_class or SchemaLoader)( self, locations, use_fallback, use_xpath3 ) @property def schemas(self) -> set[SchemaType]: return self._schemas @property def owned_schemas(self) -> set[SchemaType]: """Returns a set of the registered schemas owned by this instance.""" return {s for s in self._schemas if s.maps is self} @property def parent(self) -> Optional[SchemaType]: return self._parent def __repr__(self) -> str: return '%s(validator=%r)' % (self.__class__.__name__, self.validator) def __setattr__(self, name: str, value: Any) -> None: if hasattr(self, name): if name == '_built': self.__dict__.clear() elif name != '_parent': msg = _("can't change attribute {!r} of a global maps instance") raise XMLSchemaAttributeError(msg.format(name)) super().__setattr__(name, value) def __len__(self) -> int: return len(self._schemas) def __iter__(self) -> Iterator[SchemaType]: yield from self._schemas def __contains__(self, obj: object) -> bool: return obj in self._schemas def __getstate__(self) -> dict[str, Any]: return {a: getattr(self, a) for a in self._mro_slots() if a != '_build_lock'} def __setstate__(self, state: dict[str, Any]) -> None: for attr, value in state.items(): object.__setattr__(self, attr, value) self._build_lock = threading.Lock() def __copy__(self) -> 'XsdGlobals': other = type(self)( validator=copy.copy(self.validator), parent=self._parent, loader_class=self.loader.__class__, ) other.loader.__dict__.update(self.loader.__dict__) other.loader.locations = self.loader.locations.copy() other.loader.missing_locations.update(self.loader.missing_locations) other.validator.maps = other for schema in self._schemas: if schema.maps is self and schema is not self.validator: copy.copy(schema).maps = other other.clear() return other copy = __copy__ def lookup(self, tag: str, qname: str) -> SchemaGlobalType: """ General lookup method for XSD global components. :param tag: the expanded QName of the XSD the global declaration/definition \ (e.g. '{http://www.w3.org/2001/XMLSchema}element'), that is used to select \ the global map for lookup. :param qname: the expanded QName of the component to be looked-up. :returns: an XSD global component. :raises: an XMLSchemaValueError if the *tag* argument is not appropriate for a global \ component, an XMLSchemaKeyError if the *qname* argument is not found in the global map. """ try: global_map = GLOBAL_MAP_ATTRIBUTE[tag](self) except KeyError: msg = _("wrong tag {!r} for an XSD global definition/declaration") raise XMLSchemaValueError(msg.format(tag)) from None else: return cast(SchemaGlobalType, global_map[qname]) @deprecated('5.0', alt='use item lookup on notations instead') def lookup_notation(self, qname: str) -> XsdNotation: return self.notations[qname] @deprecated('5.0', alt='use item lookup on types instead') def lookup_type(self, qname: str) -> BaseXsdType: return self.types[qname] @deprecated('5.0', alt='use item lookup on attributes instead') def lookup_attribute(self, qname: str) -> XsdAttribute: return self.attributes[qname] @deprecated('5.0', alt='use item lookup on attribute_groups instead') def lookup_attribute_group(self, qname: str) -> XsdAttributeGroup: return self.attribute_groups[qname] @deprecated('5.0', alt='use item lookup on groups instead') def lookup_group(self, qname: str) -> XsdGroup: return self.groups[qname] @deprecated('5.0', alt='use item lookup on elements instead') def lookup_element(self, qname: str) -> XsdElement: return self.elements[qname] def get_instance_type(self, type_name: str, base_type: BaseXsdType, namespaces: NsmapType) -> BaseXsdType: """ Returns the instance XSI type from global maps, validating it with the reference base type. :param type_name: the XSI type attribute value, a QName in prefixed format. :param base_type: the XSD from which the instance type has to be derived. :param namespaces: a mapping from prefixes to namespaces. """ if isinstance(base_type, XsdComplexType) and nm.XSI_TYPE in base_type.attributes: xsd_attribute = cast(XsdAttribute, base_type.attributes[nm.XSI_TYPE]) xsd_attribute.validate(type_name) extended_name = get_extended_qname(type_name, namespaces) xsi_type = self.types[extended_name] if xsi_type.is_derived(base_type): return xsi_type elif isinstance(base_type, XsdSimpleType) and \ base_type.is_union() and not base_type.facets: # Can be valid only if the union doesn't have facets, see: # https://www.w3.org/Bugs/Public/show_bug.cgi?id=4065 if isinstance(base_type, XsdAtomicRestriction) and \ isinstance(base_type.primitive_type, XsdUnion): if xsi_type in base_type.primitive_type.member_types: return xsi_type elif isinstance(base_type, XsdUnion): if xsi_type in base_type.member_types: return xsi_type msg = _("{0!r} cannot substitute {1!r}") raise XMLSchemaTypeError(msg.format(xsi_type, base_type)) @property def built(self) -> bool: return self._built @cached_property def validation_attempted(self) -> str: if not any(m for m in self.global_maps): return 'none' elif not self._built or any(m.total_staged for m in self.global_maps): return 'partial' else: return 'full' @cached_property def validity(self) -> str: if self.validation == 'skip': return 'notKnown' elif any(s.errors for s in self._schemas) or \ any(c.errors for c in self.iter_components()): return 'invalid' elif not self._built or any(m.total_staged for m in self.global_maps): return 'notKnown' else: return 'valid' @validator_property def xpath_constructors(self) -> dict[str, Type[XPathToken]]: if not self._built: return {} xpath_parser = self.loader.xpath_parser_class() xpath_parser.schema = self.validator.xpath_proxy constructors: dict[str, Type[XPathToken]] = {} for name, xsd_type in self.types.items(): if isinstance(xsd_type, XsdAtomic) and \ not isinstance(xsd_type, XsdAtomicBuiltin) and \ name not in (nm.XSD_ANY_ATOMIC_TYPE, nm.XSD_NOTATION): constructors[name] = xpath_parser.schema_constructor(name) return constructors @property def use_meta(self) -> bool: return self.validator.meta_schema is None or \ self.validator.meta_schema in self._schemas @property def unbuilt(self) -> list[XsdComponent]: """Property that returns a list with unbuilt components.""" return [c for c in self.iter_components() if not c.built] @property def xsd_version(self) -> str: return self.validator.XSD_VERSION @property def all_errors(self) -> list[XMLSchemaParseError]: return [e for s in self._schemas for e in s.all_errors] @property def total_errors(self) -> int: return sum(s.total_errors for s in self._schemas) def create_bindings(self, *bases: Type[Any], **attrs: Any) -> None: """Creates data object bindings for the XSD elements of built schemas.""" for xsd_element in self.iter_components(xsd_classes=XsdElement): assert isinstance(xsd_element, XsdElement) if xsd_element.target_namespace != nm.XSD_NAMESPACE: xsd_element.get_binding(*bases, replace_existing=True, **attrs) def clear_bindings(self) -> None: for xsd_element in self.iter_components(xsd_classes=XsdElement): assert isinstance(xsd_element, XsdElement) xsd_element.binding = None def iter_components(self, xsd_classes: Optional[ComponentClassType] = None) \ -> Iterator[XsdComponent]: """Creates an iterator for the XSD components of built schemas.""" for xsd_global in self.global_maps.iter_globals(): yield from xsd_global.iter_components(xsd_classes) def iter_globals(self) -> Iterator[SchemaGlobalType]: """Creates an iterator for the built XSD global components.""" return self.global_maps.iter_globals() def iter_staged(self) -> Iterator[StagedItemType]: """Creates an iterator for the unbuilt XSD global components.""" return self.global_maps.iter_staged() def iter_schemas(self) -> Iterator[SchemaType]: """Creates an iterator for the registered schemas.""" yield from self._schemas def iter_ancestors(self) -> Iterator[SchemaType]: ancestors: list[SchemaType] = [] parent = self._parent while parent is not None: ancestors.append(parent) parent = parent.maps.parent yield from reversed(ancestors) def get_schema(self, namespace: Optional[str] = None, source: Optional[SchemaSourceType] = None, base_url: Optional[str] = None) -> Optional[SchemaType]: schemas: Optional[list[SchemaType]] if namespace is not None: schemas = self.namespaces.get(namespace) elif isinstance(source, XMLResource): namespace = source.root.get('targetNamespace', '') schemas = self.namespaces.get(namespace) elif source is None: return None else: schemas = list(self._schemas) if schemas is not None: if source is None: return schemas[0] elif (url := get_url(source)) is not None: url = normalize_url(url, base_url) for schema in schemas: if schema.source.match_location(url): return schema elif isinstance(source, XMLResource): for schema in schemas: if schema.source.source is source.source: return schema else: for schema in schemas: if schema.source.source is source: return schema return None def register(self, schema: SchemaType) -> None: """Registers an XMLSchema instance.""" if schema in self._schemas: return namespace = schema.target_namespace source = schema.source.url or schema.source ns_schemas = self.namespaces.get(namespace) if ns_schemas is None: self.namespaces[namespace] = [schema] self._schemas.add(schema) elif ns_schemas[0].maps is not self: self._schemas.add(schema) ns_schemas.append(schema) schema.maps = self self.merge(ancestor=ns_schemas[0].maps.validator) elif self.get_schema(namespace, source) is None: self._schemas.add(schema) ns_schemas.append(schema) else: source_ref = schema.source.url or type(schema.source) raise XMLSchemaValueError( f"another schema loaded from {source_ref!r} is already registered" ) self._built = False def merge(self, ancestor: SchemaType) -> None: """Merge the global maps until to a specific ancestor.""" self.clear() for validator in dropwhile(lambda x: x is not ancestor, self.iter_ancestors()): maps = validator.maps for schema in maps._schemas: if schema.maps is maps: namespace = schema.target_namespace self._schemas.remove(schema) k = self.namespaces[namespace].index(schema) schema = copy.copy(schema) self._schemas.add(schema) self.namespaces[namespace][k] = schema schema.maps = self self._parent = ancestor.maps._parent def clear(self, remove_schemas: bool = False) -> None: """ Clears the instance maps and schemas. :param remove_schemas: removes also the schema instances, keeping only the \ validator that created the global maps instance and schemas and namespaces \ inherited from ancestors. """ self.global_maps.clear() self.substitution_groups.clear() self.identities.clear() for schema in self._schemas: if schema.maps is self: schema.clear() # clear XPath and component lazy properties if remove_schemas: self._schemas.clear() self.namespaces.clear() for ancestor in self.iter_ancestors(): self._schemas.update(ancestor.maps._schemas) self.namespaces.update(ancestor.maps.namespaces) self.register(self.validator) if self.validator.includes: self.validator.includes.clear() self._built = False def build(self) -> None: """ Build the maps of XSD global definitions/declarations. The global maps are updated adding and building the globals of not built registered schemas. """ if self._built: return with self._build_lock: if self._built: return self.check_loaded_schemas() self.clear() for ancestor in self.iter_ancestors(): ancestor.maps.build() self.global_maps.update(ancestor.maps.global_maps) self.substitution_groups.update(ancestor.maps.substitution_groups) self.identities.update(ancestor.maps.identities) # Have to respect the insertion order for redefinitions/overrides schemas = [s for ns_schemas in self.namespaces.values() for s in ns_schemas if s.maps is self] self.global_maps.load(schemas) self.types.build_builtins(self.validator) self.global_maps.build(schemas) self.check(schemas) self._built = True for s in schemas: s.clear() self.check_validator() @contextmanager def protect_status(self, reraise: bool = True) -> Iterator['XsdGlobals']: """Context manager for set a restore point in case of error.""" schemas = self._schemas.copy() namespaces = self.namespaces.copy() global_maps = self.global_maps.copy() substitution_groups = self.substitution_groups.copy() identities = self.identities.copy() built = self._built try: yield self except XMLSchemaException: self.clear() self._schemas.clear() self.namespaces.clear() self._schemas.update(schemas) self.namespaces.update(namespaces) self.global_maps.update(global_maps) self.substitution_groups.update(substitution_groups) self.identities.update(identities) self._built = built if reraise: raise def check_loaded_schemas(self) -> None: """Checks the coherence of schema registrations.""" if self.validator not in self._schemas: raise XMLSchemaValueError(_('global maps main validator is not registered')) if self._parent is None: self.validator.create_meta_schema(global_maps=self) total = 0 for namespace, schemas in self.namespaces.items(): total += len(schemas) for s in schemas: if s.target_namespace != namespace: msg = _('schema {!r} does not belong to namespace {!r}') raise XMLSchemaNamespaceError(msg.format(s, namespace)) elif s not in self._schemas: msg = _('schema {!r} does not belong to {!r}') raise XMLSchemaNamespaceError(msg.format(s, self)) if len(self._schemas) != total: raise XMLSchemaValueError( _('registered schemas do not match namespace mapped schemas') ) def check(self, schemas: Optional[Iterable[SchemaType]] = None) -> None: """ Checks the components of provided schemas. Used after the build of the global maps. For default checks all schemas and raises an exception at first error. :param schemas: optional argument with the set of the schemas to check. :raise: XMLSchemaParseError """ if schemas is None: schemas = {s for s in self._schemas if s.maps is self} # Checks substitution groups circularity for qname in self.substitution_groups: xsd_element = self.elements[qname] if any(e is xsd_element for e in xsd_element.iter_substitutes()): msg = _("circularity found for substitution group with head element {}") xsd_element.parse_error(msg.format(xsd_element)) # Check redefined global groups restrictions for group in self.groups.values(): assert isinstance(group, XsdGroup), _("global group not built!") if group.schema not in schemas or group.redefine is None: continue while group.redefine is not None: if not any(isinstance(e, XsdGroup) and e.name == group.name for e in group) \ and not group.is_restriction(group.redefine): msg = _("the redefined group is an illegal restriction") group.parse_error(msg) group = group.redefine # Check complex content types models restrictions for xsd_global in filter(lambda x: x.schema in schemas, self.iter_globals()): xsd_type: Any for xsd_type in xsd_global.iter_components(XsdComplexType): if not isinstance(xsd_type.content, XsdGroup): continue if xsd_type.derivation == 'restriction': base_type = xsd_type.base_type if base_type and base_type.name != nm.XSD_ANY_TYPE and base_type.is_complex(): if not xsd_type.content.is_restriction(base_type.content): msg = _("the derived group is an illegal restriction") xsd_type.parse_error(msg) if base_type.is_complex() and not base_type.open_content and \ xsd_type.open_content and xsd_type.open_content.mode != 'none': _group = xsd_type.schema.builders.create_any_content_group( parent=xsd_type, any_element=xsd_type.open_content.any_element ) if not _group.is_restriction(base_type.content): msg = _("restriction has an open content but base type has not") _group.parse_error(msg) try: check_model(xsd_type.content) except XMLSchemaModelDepthError: msg = _("can't verify the content model of {!r} " "due to exceeding of maximum recursion depth") xsd_type.schema.warnings.append(msg.format(xsd_type)) warnings.warn(msg, XMLSchemaWarning, stacklevel=4) except XMLSchemaModelError as err: if self.validation == 'strict': raise xsd_type.errors.append(err)