path: root/django/core/serializers/xml_serializer.py

"""
XML serializer.
"""

import json
from contextlib import contextmanager
from xml.dom import minidom, pulldom
from xml.sax import handler
from xml.sax.expatreader import ExpatParser as _ExpatParser

from django.apps import apps
from django.conf import settings
from django.core.exceptions import ObjectDoesNotExist, SuspiciousOperation
from django.core.serializers import base
from django.db import DEFAULT_DB_ALIAS, models
from django.utils.xmlutils import SimplerXMLGenerator, UnserializableContentError


@contextmanager
def fast_cache_clearing():
    """Workaround for performance issues in minidom document checks.

    Speeds up repeated DOM operations by skipping unnecessary full traversal
    of the DOM tree.
    """
    module_helper_was_lambda = False
    if original_fn := getattr(minidom, "_in_document", None):
        module_helper_was_lambda = original_fn.__name__ == "<lambda>"
        if not module_helper_was_lambda:
            minidom._in_document = lambda node: bool(node.ownerDocument)
    try:
        yield
    finally:
        if original_fn and not module_helper_was_lambda:
            minidom._in_document = original_fn


class Serializer(base.Serializer):
    """Serialize a QuerySet to XML."""

    def indent(self, level):
        if self.options.get("indent") is not None:
            self.xml.ignorableWhitespace(
                "\n" + " " * self.options.get("indent") * level
            )

    def start_serialization(self):
        """
        Start serialization -- open the XML document and the root element.
        """
        # Increment the indent_level before each startElement() and decrement
        # it following each endElement(). If the closing tag should appear on
        # its own line, use self.indent(self.indent_level) before endElement().
        self.indent_level = 0
        self.xml = SimplerXMLGenerator(
            self.stream, self.options.get("encoding", settings.DEFAULT_CHARSET)
        )
        self.xml.startDocument()
        self.xml.startElement("django-objects", {"version": "1.0"})

    def end_serialization(self):
        """
        End serialization -- end the document.
        """
        self.indent(self.indent_level)
        self.xml.endElement("django-objects")
        self.xml.endDocument()

    def start_object(self, obj):
        """
        Called as each object is handled.
        """
        if not hasattr(obj, "_meta"):
            raise base.SerializationError(
                "Non-model object (%s) encountered during serialization" % type(obj)
            )

        self.indent_level += 1
        self.indent(self.indent_level)
        attrs = {"model": str(obj._meta)}
        if not self.use_natural_primary_keys or not self._resolve_natural_key(obj):
            obj_pk = obj.pk
            if obj_pk is not None:
                attrs["pk"] = obj._meta.pk.value_to_string(obj)

        self.xml.startElement("object", attrs)

    def end_object(self, obj):
        """
        Called after handling all fields for an object.
        """
        self.indent(self.indent_level)
        self.xml.endElement("object")
        self.indent_level -= 1

    def handle_field(self, obj, field):
        """
        Handle each field on an object (except for ForeignKeys and
        ManyToManyFields).
        """
        self.indent_level += 1
        self.indent(self.indent_level)
        self.xml.startElement(
            "field",
            {
                "name": field.name,
                "type": field.get_internal_type(),
            },
        )

        # Get a "string version" of the object's data.
        if getattr(obj, field.name) is not None:
            value = field.value_to_string(obj)
            if field.get_internal_type() == "JSONField":
                # Dump value since JSONField.value_to_string() doesn't output
                # strings.
                value = json.dumps(value, cls=field.encoder)
            try:
                self.xml.characters(value)
            except UnserializableContentError:
                raise ValueError(
                    "%s.%s (pk:%s) contains unserializable characters"
                    % (obj.__class__.__name__, field.name, obj.pk)
                )
        else:
            self.xml.addQuickElement("None")

        self.xml.endElement("field")
        self.indent_level -= 1

    def handle_fk_field(self, obj, field):
        """
        Handle a ForeignKey (they need to be treated slightly
        differently from regular fields).
        """
        self._start_relational_field(field)
        related_att = getattr(obj, field.attname)
        if related_att is not None:
            if self.use_natural_foreign_keys and (
                natural_key_value := self._resolve_fk_natural_key(obj, field)
            ):
                # Iterable natural keys are rolled out as subelements
                for key_value in natural_key_value:
                    self.xml.startElement("natural", {})
                    if key_value is None:
                        self.xml.addQuickElement("None")
                    else:
                        self.xml.characters(str(key_value))
                    self.xml.endElement("natural")
            else:
                self.xml.characters(str(related_att))
        else:
            self.xml.addQuickElement("None")
        self.xml.endElement("field")
        self.indent_level -= 1

    def handle_m2m_field(self, obj, field):
        """
        Handle a ManyToManyField. Related objects are only serialized as
        references to the object's PK (i.e. the related *data* is not dumped,
        just the relation).
        """
        if field.remote_field.through._meta.auto_created:
            self._start_relational_field(field)
            if self.use_natural_foreign_keys and self._model_supports_natural_key(
                field.remote_field.model
            ):
                # If the objects in the m2m have a natural key, use it
                def handle_m2m(value):
                    if natural := self._resolve_natural_key(value):
                        # Iterable natural keys are rolled out as subelements
                        self.xml.startElement("object", {})
                        for key_value in natural:
                            self.xml.startElement("natural", {})
                            if key_value is None:
                                self.xml.addQuickElement("None")
                            else:
                                self.xml.characters(str(key_value))
                            self.xml.endElement("natural")
                        self.xml.endElement("object")
                    else:
                        self.xml.addQuickElement("object", attrs={"pk": str(value.pk)})

                def queryset_iterator(obj, field):
                    attr = getattr(obj, field.name)
                    chunk_size = (
                        2000 if getattr(attr, "prefetch_cache_name", None) else None
                    )
                    query_set = attr.all()
                    if not query_set.totally_ordered:
                        current_ordering = (
                            query_set.query.order_by
                            or query_set.model._meta.ordering
                            or []
                        )
                        query_set = query_set.order_by(*current_ordering, "pk")
                    return query_set.iterator(chunk_size)

            else:

                def handle_m2m(value):
                    # Put each object on its own line.
                    self.indent(self.indent_level + 1)
                    self.xml.addQuickElement("object", attrs={"pk": str(value.pk)})

                def queryset_iterator(obj, field):
                    query_set = getattr(obj, field.name).select_related(None).only("pk")
                    if not query_set.totally_ordered:
                        current_ordering = (
                            query_set.query.order_by
                            or query_set.model._meta.ordering
                            or []
                        )
                        query_set = query_set.order_by(*current_ordering, "pk")
                    chunk_size = 2000 if query_set._prefetch_related_lookups else None
                    return query_set.iterator(chunk_size=chunk_size)

            m2m_iter = getattr(obj, "_prefetched_objects_cache", {}).get(
                field.name,
                queryset_iterator(obj, field),
            )
            relobj = None
            for relobj in m2m_iter:
                handle_m2m(relobj)
            if relobj:
                # If there are related objects (which appear each on their own
                # line), put the closing </field> on the next line.
                self.indent(self.indent_level)
            self.xml.endElement("field")
            self.indent_level -= 1

    def _start_relational_field(self, field):
        """Output the <field> element for relational fields."""
        self.indent_level += 1
        self.indent(self.indent_level)
        self.xml.startElement(
            "field",
            {
                "name": field.name,
                "rel": field.remote_field.__class__.__name__,
                "to": str(field.remote_field.model._meta),
            },
        )


class Deserializer(base.Deserializer):
    """Deserialize XML."""

    def __init__(
        self,
        stream_or_string,
        *,
        using=DEFAULT_DB_ALIAS,
        ignorenonexistent=False,
        **options,
    ):
        super().__init__(stream_or_string, **options)
        self.handle_forward_references = options.pop("handle_forward_references", False)
        self.event_stream = pulldom.parse(self.stream, self._make_parser())
        self.db = using
        self.ignore = ignorenonexistent

    def _make_parser(self):
        """Create a hardened XML parser (no custom/external entities)."""
        return DefusedExpatParser()

    def __next__(self):
        for event, node in self.event_stream:
            if event == "START_ELEMENT" and node.nodeName == "object":
                with fast_cache_clearing():
                    self.event_stream.expandNode(node)
                return self._handle_object(node)
        raise StopIteration

    def _handle_object(self, node):
        """Convert an <object> node to a DeserializedObject."""
        # Look up the model using the model loading mechanism. If this fails,
        # bail.
        Model = self._get_model_from_node(node, "model")

        # Start building a data dictionary from the object.
        data = {}
        if node.hasAttribute("pk"):
            data[Model._meta.pk.attname] = Model._meta.pk.to_python(
                node.getAttribute("pk")
            )

        # Also start building a dict of m2m data (this is saved as
        # {m2m_accessor_attribute : [list_of_related_objects]})
        m2m_data = {}
        deferred_fields = {}

        field_names = {f.name for f in Model._meta.get_fields()}
        # Deserialize each field.
        for field_node in getChildrenByTagName(node, "field"):
            # If the field is missing the name attribute, bail (are you
            # sensing a pattern here?)
            field_name = field_node.getAttribute("name")
            if not field_name:
                raise base.DeserializationError(
                    "<field> node is missing the 'name' attribute"
                )

            # Get the field from the Model. This will raise a
            # FieldDoesNotExist if, well, the field doesn't exist, which will
            # be propagated correctly unless ignorenonexistent=True is used.
            if self.ignore and field_name not in field_names:
                continue
            field = Model._meta.get_field(field_name)

            # As is usually the case, relation fields get the special
            # treatment.
            if field.remote_field and isinstance(
                field.remote_field, models.ManyToManyRel
            ):
                value = self._handle_m2m_field_node(field_node, field)
                if value == base.DEFER_FIELD:
                    deferred_fields[field] = [
                        [
                            (
                                None
                                if getChildrenByTagName(nat_node, "None")
                                else getInnerText(nat_node).strip()
                            )
                            for nat_node in getChildrenByTagName(obj_node, "natural")
                        ]
                        for obj_node in getChildrenByTagName(field_node, "object")
                    ]
                else:
                    m2m_data[field.name] = value
            elif field.remote_field and isinstance(
                field.remote_field, models.ManyToOneRel
            ):
                value = self._handle_fk_field_node(field_node, field)
                if value == base.DEFER_FIELD:
                    deferred_fields[field] = [
                        (
                            None
                            if getChildrenByTagName(k, "None")
                            else getInnerText(k).strip()
                        )
                        for k in getChildrenByTagName(field_node, "natural")
                    ]
                else:
                    data[field.attname] = value
            else:
                if getChildrenByTagName(field_node, "None"):
                    value = None
                else:
                    value = field.to_python(getInnerText(field_node).strip())
                    # Load value since JSONField.to_python() outputs strings.
                    if field.get_internal_type() == "JSONField":
                        value = json.loads(value, cls=field.decoder)
                data[field.name] = value

        obj = base.build_instance(Model, data, self.db)

        # Return a DeserializedObject so that the m2m data has a place to live.
        return base.DeserializedObject(obj, m2m_data, deferred_fields)

    def _handle_fk_field_node(self, node, field):
        """
        Handle a <field> node for a ForeignKey
        """
        # Check if there is a child node named 'None', returning None if so.
        natural_keys = getChildrenByTagName(node, "natural")
        if getChildrenByTagName(node, "None") and not natural_keys:
            return None
        else:
            model = field.remote_field.model
            if hasattr(model._default_manager, "get_by_natural_key"):
                if natural_keys:
                    # If there are 'natural' subelements, it must be a natural
                    # key
                    field_value = []
                    for k in natural_keys:
                        if getChildrenByTagName(k, "None"):
                            field_value.append(None)
                        else:
                            field_value.append(getInnerText(k).strip())
                    try:
                        obj = model._default_manager.db_manager(
                            self.db
                        ).get_by_natural_key(*field_value)
                    except ObjectDoesNotExist:
                        if self.handle_forward_references:
                            return base.DEFER_FIELD
                        else:
                            raise
                    obj_pk = getattr(obj, field.remote_field.field_name)
                    # If this is a natural foreign key to an object that
                    # has a FK/O2O as the foreign key, use the FK value
                    if field.remote_field.model._meta.pk.remote_field:
                        obj_pk = obj_pk.pk
                else:
                    # Otherwise, treat like a normal PK
                    field_value = getInnerText(node).strip()
                    obj_pk = model._meta.get_field(
                        field.remote_field.field_name
                    ).to_python(field_value)
                return obj_pk
            else:
                field_value = getInnerText(node).strip()
                return model._meta.get_field(field.remote_field.field_name).to_python(
                    field_value
                )

    def _handle_m2m_field_node(self, node, field):
        """
        Handle a <field> node for a ManyToManyField.
        """
        model = field.remote_field.model
        default_manager = model._default_manager
        if hasattr(default_manager, "get_by_natural_key"):

            def m2m_convert(n):
                keys = getChildrenByTagName(n, "natural")
                if keys:
                    # If there are 'natural' subelements, it must be a natural
                    # key
                    field_value = []
                    for k in keys:
                        if getChildrenByTagName(k, "None"):
                            field_value.append(None)
                        else:
                            field_value.append(getInnerText(k).strip())
                    obj_pk = (
                        default_manager.db_manager(self.db)
                        .get_by_natural_key(*field_value)
                        .pk
                    )
                else:
                    # Otherwise, treat like a normal PK value.
                    obj_pk = model._meta.pk.to_python(n.getAttribute("pk"))
                return obj_pk

        else:

            def m2m_convert(n):
                return model._meta.pk.to_python(n.getAttribute("pk"))

        values = []
        try:
            for c in getChildrenByTagName(node, "object"):
                values.append(m2m_convert(c))
        except SuspiciousOperation:
            raise
        except Exception as e:
            if isinstance(e, ObjectDoesNotExist) and self.handle_forward_references:
                return base.DEFER_FIELD
            else:
                raise base.M2MDeserializationError(e, c)
        else:
            return values

    def _get_model_from_node(self, node, attr):
        """
        Look up a model from a <object model=...> or a <field rel=... to=...>
        node.
        """
        model_identifier = node.getAttribute(attr)
        if not model_identifier:
            raise base.DeserializationError(
                "<%s> node is missing the required '%s' attribute"
                % (node.nodeName, attr)
            )
        try:
            return apps.get_model(model_identifier)
        except (LookupError, TypeError):
            raise base.DeserializationError(
                "<%s> node has invalid model identifier: '%s'"
                % (node.nodeName, model_identifier)
            )


def check_element_type(element):
    if element.childNodes:
        raise SuspiciousOperation(f"Unexpected element: {element.tagName!r}")
    return element.nodeType in (element.TEXT_NODE, element.CDATA_SECTION_NODE)


def getChildrenByTagName(node, tag_name):
    """
    Like Element.getElementsByTagName() but return only direct children.

    Element.getElementsByTagName() searches all descendants (direct children,
    children's children, etc.). This prevents correct deserialization of
    third-party nested fields. For example:

      <field name="author" type="EmbeddedModelField">
        <object model="app.Model">
          <field name="id" type="..."><None></None></field>
    """
    return [
        n
        for n in node.childNodes
        if n.nodeType == node.ELEMENT_NODE and n.tagName == tag_name
    ]


def getInnerText(node):
    return "".join(
        [child.data for child in node.childNodes if check_element_type(child)]
    )


# Below code based on Christian Heimes' defusedxml


class DefusedExpatParser(_ExpatParser):
    """
    An expat parser hardened against XML bomb attacks.

    Forbid DTDs, external entity references
    """

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.setFeature(handler.feature_external_ges, False)
        self.setFeature(handler.feature_external_pes, False)

    def start_doctype_decl(self, name, sysid, pubid, has_internal_subset):
        raise DTDForbidden(name, sysid, pubid)

    def entity_decl(
        self, name, is_parameter_entity, value, base, sysid, pubid, notation_name
    ):
        raise EntitiesForbidden(name, value, base, sysid, pubid, notation_name)

    def unparsed_entity_decl(self, name, base, sysid, pubid, notation_name):
        # expat 1.2
        raise EntitiesForbidden(name, None, base, sysid, pubid, notation_name)

    def external_entity_ref_handler(self, context, base, sysid, pubid):
        raise ExternalReferenceForbidden(context, base, sysid, pubid)

    def reset(self):
        _ExpatParser.reset(self)
        parser = self._parser
        parser.StartDoctypeDeclHandler = self.start_doctype_decl
        parser.EntityDeclHandler = self.entity_decl
        parser.UnparsedEntityDeclHandler = self.unparsed_entity_decl
        parser.ExternalEntityRefHandler = self.external_entity_ref_handler


class DefusedXmlException(ValueError):
    """Base exception."""

    def __repr__(self):
        return str(self)


class DTDForbidden(DefusedXmlException):
    """Document type definition is forbidden."""

    def __init__(self, name, sysid, pubid):
        super().__init__()
        self.name = name
        self.sysid = sysid
        self.pubid = pubid

    def __str__(self):
        tpl = "DTDForbidden(name='{}', system_id={!r}, public_id={!r})"
        return tpl.format(self.name, self.sysid, self.pubid)


class EntitiesForbidden(DefusedXmlException):
    """Entity definition is forbidden."""

    def __init__(self, name, value, base, sysid, pubid, notation_name):
        super().__init__()
        self.name = name
        self.value = value
        self.base = base
        self.sysid = sysid
        self.pubid = pubid
        self.notation_name = notation_name

    def __str__(self):
        tpl = "EntitiesForbidden(name='{}', system_id={!r}, public_id={!r})"
        return tpl.format(self.name, self.sysid, self.pubid)


class ExternalReferenceForbidden(DefusedXmlException):
    """Resolving an external reference is forbidden."""

    def __init__(self, context, base, sysid, pubid):
        super().__init__()
        self.context = context
        self.base = base
        self.sysid = sysid
        self.pubid = pubid

    def __str__(self):
        tpl = "ExternalReferenceForbidden(system_id='{}', public_id={})"
        return tpl.format(self.sysid, self.pubid)