Harden URL reads and XML serialization

.github/dependabot.yml

@@ -4,3 +4,11 @@ updates:
     directory: /
     schedule:
       interval: weekly
+  - package-ecosystem: pip
+    directory: /
+    schedule:
+      interval: weekly
+  - package-ecosystem: pip
+    directory: /docs
+    schedule:
+      interval: weekly

README.rst

@@ -239,6 +239,18 @@ You can use the json2xml library in the following ways:
     data = readfromjson("examples/licht.json")
     print(json2xml.Json2xml(data).to_xml())
 
+URL reads accept only credential-free HTTP(S), reject redirects and non-public
+destinations by default, and stop after 10 MiB of decoded content. Trusted
+library callers can opt into a private endpoint or choose a smaller limit:
+
+.. code-block:: python
+
+    data = readfromurl(
+        "http://127.0.0.1:8000/data.json",
+        allow_private_networks=True,
+        max_response_bytes=1024 * 1024,
+    )
+
 
 Custom Wrappers and Indentation
 ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^

docs/usage.rst

@@ -32,6 +32,19 @@ Here's how to use each method:
     data = readfromjson("examples/licht.json")
     print(json2xml.Json2xml(data).to_xml())
 
+URL reads accept only credential-free HTTP(S), reject redirects and non-public
+destinations by default, and stop after 10 MiB of decoded content. Trusted
+library callers can explicitly allow a private endpoint while retaining a
+chosen response limit:
+
+.. code-block:: python
+
+    data = readfromurl(
+        "http://127.0.0.1:8000/data.json",
+        allow_private_networks=True,
+        max_response_bytes=1024 * 1024,
+    )
+
 
 Real-world Examples
 -------------------

json2xml/dicttoxml.py

@@ -125,6 +125,28 @@ def get_xml_type(val: Any) -> str:
     return type(val).__name__
 
 
+# @lat: [[behavior#XML output safety]]
+def _validate_xml_chars(value: str) -> None:
+    """Reject characters excluded from the XML 1.0 Char production."""
+    # XML 1.0 character production: https://www.w3.org/TR/xml/#charsets
+    # isprintable() keeps the common path in C; only strings containing a
+    # non-printable character need the explicit code-point boundary checks.
+    if not value or value.isprintable():
+        return
+
+    for character in value:
+        codepoint = ord(character)
+        is_forbidden_control = (
+            codepoint < 0x20 and codepoint not in (0x09, 0x0A, 0x0D)
+        )
+        is_surrogate = 0xD800 <= codepoint <= 0xDFFF
+        is_bmp_noncharacter = codepoint in (0xFFFE, 0xFFFF)
+        if is_forbidden_control or is_surrogate or is_bmp_noncharacter:
+            raise ValueError(
+                f"Character U+{codepoint:04X} is not allowed in XML 1.0"
+            )
+
+
 def escape_xml(s: str | int | float | numbers.Number | None) -> str:
     """
     Escape a string for use in XML.
@@ -135,15 +157,15 @@ def escape_xml(s: str | int | float | numbers.Number | None) -> str:
     Returns:
         str: The escaped string.
     """
-    if isinstance(s, str):
-        if _XML_ESCAPE_CHARS.isdisjoint(s):
-            return s
-        s = s.replace("&", "&amp;")
-        s = s.replace('"', "&quot;")
-        s = s.replace("'", "&apos;")
-        s = s.replace("<", "&lt;")
-        s = s.replace(">", "&gt;")
-    return str(s)
+    value = s if isinstance(s, str) else str(s)
+    _validate_xml_chars(value)
+    if _XML_ESCAPE_CHARS.isdisjoint(value):
+        return value
+    value = value.replace("&", "&amp;")
+    value = value.replace('"', "&quot;")
+    value = value.replace("'", "&apos;")
+    value = value.replace("<", "&lt;")
+    return value.replace(">", "&gt;")
 
 
 def make_attrstring(attr: dict[str, Any]) -> str:
@@ -161,7 +183,7 @@ def make_attrstring(attr: dict[str, Any]) -> str:
     if len(attr) == 1:
         key, val = next(iter(attr.items()))
         if key == "type":
-            return f' type="{val}"'
+            return f' type="{escape_xml(val)}"'
         validate_xml_attr_names(attr)
         return f' {key}="{escape_xml(val)}"'
     validate_xml_attr_names(attr)
@@ -263,8 +285,10 @@ def make_valid_xml_name(key: str, attr: dict[str, Any]) -> tuple[str, dict[str,
 
 def wrap_cdata(s: str | int | float | numbers.Number) -> str:
     """Wraps a string into CDATA sections"""
-    s = str(s).replace("]]>", "]]]]><![CDATA[>")
-    return "<![CDATA[" + s + "]]>"
+    value = str(s)
+    _validate_xml_chars(value)
+    value = value.replace("]]>", "]]]]><![CDATA[>")
+    return "<![CDATA[" + value + "]]>"
 
 
 def default_item_func(parent: str) -> str:
@@ -1113,6 +1137,7 @@ def render(self) -> bytes:
         return output.to_bytes()
 
 
+# @lat: [[behavior#XML output safety]]
 class _NamespaceFormatter:
     """Keep namespace emission and schema-attribute quirks in one place."""
 
@@ -1122,21 +1147,38 @@ def format(xml_namespaces: dict[str, Any] | None) -> str:
             return ""
 
         namespace_parts: list[str] = []
-        for prefix in xml_namespaces:
+        for prefix, namespace_value in xml_namespaces.items():
             if prefix == "xsi":
-                for schema_att in xml_namespaces[prefix]:
+                if not isinstance(namespace_value, dict):
+                    raise ValueError("The xsi namespace value must be a mapping")
+                if (
+                    "schemaLocation" in namespace_value
+                    and "schemaInstance" not in namespace_value
+                ):
+                    raise ValueError(
+                        "xsi schemaLocation requires a schemaInstance namespace"
+                    )
+                for schema_att in namespace_value:
                     if schema_att == "schemaInstance":
                         namespace_parts.append(
-                            f' xmlns:{prefix}="{xml_namespaces[prefix]["schemaInstance"]}"'
+                            f' xmlns:{prefix}="{escape_xml(namespace_value[schema_att])}"'
                         )
                     elif schema_att == "schemaLocation":
                         namespace_parts.append(
-                            f' xsi:{schema_att}="{xml_namespaces[prefix][schema_att]}"'
+                            f' xsi:{schema_att}="{escape_xml(namespace_value[schema_att])}"'
                         )
             elif prefix == "xmlns":
-                namespace_parts.append(f' xmlns="{xml_namespaces[prefix]}"')
+                namespace_parts.append(f' xmlns="{escape_xml(namespace_value)}"')
             else:
-                namespace_parts.append(f' xmlns:{prefix}="{xml_namespaces[prefix]}"')
+                if (
+                    not isinstance(prefix, str)
+                    or prefix.lower().startswith("xml")
+                    or not key_is_valid_xml(prefix)
+                ):
+                    raise ValueError(f"Invalid XML namespace prefix: {prefix}")
+                namespace_parts.append(
+                    f' xmlns:{prefix}="{escape_xml(namespace_value)}"'
+                )
         return "".join(namespace_parts)
 
 

json2xml/dicttoxml_fast.py

@@ -30,12 +30,29 @@
 rust_escape_xml: RustStringTransform | None = None
 rust_wrap_cdata: RustStringTransform | None = None
 
+
+def _rejects_invalid_xml(escape: RustStringTransform) -> bool:
+    """Return whether an optional backend enforces XML 1.0 characters."""
+    try:
+        escape("\x00")
+    except ValueError:
+        return True
+    except Exception:
+        return False
+    return False
+
+
 try:
     from json2xml_rs import dicttoxml as _rust_dicttoxml  # pragma: no cover
     from json2xml_rs import escape_xml_py as rust_escape_xml  # pragma: no cover
     from json2xml_rs import wrap_cdata_py as rust_wrap_cdata  # pragma: no cover
-    _use_rust = True  # pragma: no cover
-    LOG.debug("Using Rust backend for dicttoxml")  # pragma: no cover
+    if _rejects_invalid_xml(rust_escape_xml):  # pragma: no cover
+        _use_rust = True  # pragma: no cover
+        LOG.debug("Using Rust backend for dicttoxml")  # pragma: no cover
+    else:  # pragma: no cover
+        LOG.warning(  # pragma: no cover
+            "Ignoring an outdated Rust backend that permits invalid XML characters"
+        )
 except ImportError:  # pragma: no cover
     LOG.debug("Rust backend not available, using pure Python")
 

json2xml/json2xml.py

@@ -41,16 +41,19 @@ def to_xml(self) -> Any | None:
         Convert to xml using dicttoxml.dicttoxml and then pretty print it.
         """
         if self.data is not None:
-            xml_data = dicttoxml.dicttoxml(
-                self.data,
-                root=self.root,
-                custom_root=self.wrapper,
-                attr_type=self.attr_type,
-                item_wrap=self.item_wrap,
-                xpath_format=self.xpath_format,
-                cdata=self.cdata,
-                list_headers=self.list_headers,
-            )
+            try:
+                xml_data = dicttoxml.dicttoxml(
+                    self.data,
+                    root=self.root,
+                    custom_root=self.wrapper,
+                    attr_type=self.attr_type,
+                    item_wrap=self.item_wrap,
+                    xpath_format=self.xpath_format,
+                    cdata=self.cdata,
+                    list_headers=self.list_headers,
+                )
+            except ValueError as error:
+                raise InvalidDataError from error
             if self.pretty:
                 from pyexpat import ExpatError
 

json2xml/utils.py

@@ -2,13 +2,17 @@
 from __future__ import annotations
 
 import json
+import socket
+from ipaddress import ip_address
 from typing import Any
+from urllib.parse import urlsplit
 
 __lazy_modules__ = ["urllib3"]
 
 from .types import JSONValue
 
 DEFAULT_URL_TIMEOUT: Any | None = None
+DEFAULT_MAX_RESPONSE_BYTES = 10 * 1024 * 1024
 _HTTP: Any | None = None
 
 
@@ -54,25 +58,99 @@ def readfromjson(filename: str) -> JSONValue:
         raise JSONReadError("Invalid JSON File") from error
 
 
-def readfromurl(url: str, params: dict[str, str] | None = None) -> JSONValue:
-    """Load JSON data from a URL."""
+# @lat: [[behavior#URL security boundaries]]
+def _validate_url(url: str, allow_private_networks: bool) -> None:
+    """Reject URL forms that can escape the intended public HTTP boundary."""
+    try:
+        parsed = urlsplit(url)
+        port = parsed.port
+    except (TypeError, ValueError) as error:
+        raise URLReadError("URL is not valid") from error
+
+    if parsed.scheme not in {"http", "https"}:
+        raise URLReadError("URL must use HTTP or HTTPS")
+    if parsed.username is not None or parsed.password is not None:
+        raise URLReadError("URL must not contain credentials")
+    if parsed.hostname is None:
+        raise URLReadError("URL must include a hostname")
+    if allow_private_networks:
+        return
+
+    hostname = parsed.hostname
+    try:
+        addresses = {ip_address(hostname)}
+    except ValueError:
+        try:
+            address_info = socket.getaddrinfo(
+                hostname,
+                port or (443 if parsed.scheme == "https" else 80),
+                type=socket.SOCK_STREAM,
+            )
+        except OSError as error:
+            raise URLReadError("URL hostname could not be resolved") from error
+        addresses = {
+            ip_address(str(info[4][0]).split("%", 1)[0])
+            for info in address_info
+        }
+
+    if not addresses or any(not address.is_global for address in addresses):
+        raise URLReadError("URL must resolve only to a public network address")
+
+
+def readfromurl(
+    url: str,
+    params: dict[str, str] | None = None,
+    *,
+    max_response_bytes: int = DEFAULT_MAX_RESPONSE_BYTES,
+    allow_private_networks: bool = False,
+) -> JSONValue:
+    """Load bounded JSON data from a public URL.
+
+    Private-network access is available only through the explicit trusted-caller
+    opt-in. Redirects and embedded credentials are always rejected.
+    """
+    if (
+        isinstance(max_response_bytes, bool)
+        or not isinstance(max_response_bytes, int)
+        or max_response_bytes <= 0
+    ):
+        raise URLReadError("Maximum response size must be a positive integer")
+    _validate_url(url, allow_private_networks)
+
     urllib3, http, timeout = _get_http_client()
+    response = None
     try:
         response = http.request(
             "GET",
             url,
             fields=params,
             timeout=timeout,
             retries=False,
+            redirect=False,
+            preload_content=False,
         )
+        if response.status != 200:
+            raise URLReadError("URL is not returning correct response")
+
+        content_length = response.headers.get("Content-Length")
+        if content_length is not None:
+            try:
+                if int(content_length) > max_response_bytes:
+                    raise URLReadError("URL response exceeds maximum size")
+            except ValueError as error:
+                raise URLReadError("URL returned an invalid Content-Length") from error
+
+        response_data = response.read(max_response_bytes + 1, decode_content=True)
+        if len(response_data) > max_response_bytes:
+            raise URLReadError("URL response exceeds maximum size")
     except urllib3.exceptions.HTTPError as error:
         raise URLReadError("URL could not be read") from error
-
-    if response.status != 200:
-        raise URLReadError("URL is not returning correct response")
+    finally:
+        if response is not None:
+            response.close()
 
     try:
-        return json.loads(response.data.decode("utf-8"))
+        return json.loads(response_data.decode("utf-8"))
     except (UnicodeDecodeError, json.JSONDecodeError) as error:
         raise URLReadError("URL did not return valid JSON") from error
 

lat.md/architecture.md

@@ -18,11 +18,13 @@ The `dicttoxml()` entry point now normalizes options into `SerializerConfig` and
 
 The recursive serializer still streams normal and XPath serialization through [[json2xml/dicttoxml.py#_XMLWriter]] so dict and list payloads do not allocate a complete string for each nested subtree. Public helpers such as `convert_dict()` still return strings for compatibility by delegating to the same append path, while library and CLI conversions write UTF-8 bytes incrementally and return the final `bytes` object. Attribute formatting stays centralized through `make_attrstring()`, and `@attrs`/`@val` normalization stays local to dict element handling so caller-owned metadata is never mutated.
 
+Text, CDATA, custom attributes, and namespace declarations share XML 1.0 character validation. Namespace declarations additionally validate prefixes before the renderer appends them to the root element.
+
 ## Backend selection
 
 The fast-path module prefers the Rust extension when it can preserve Python semantics, and falls back to the Python serializer for unsupported features.
 
-[[json2xml/dicttoxml_fast.py#dicttoxml]] now normalizes each call into a shared conversion request and asks a tiny backend selector seam to choose Rust or Python. The Rust adapter accepts only requests whose semantics it can preserve, namely no `ids`, custom `item_func`, XML namespaces, XPath mode, root scalar payloads, or special `@` keys.
+[[json2xml/dicttoxml_fast.py#dicttoxml]] now normalizes each call into a shared conversion request and asks a tiny backend selector seam to choose Rust or Python. The Rust adapter accepts only requests whose semantics it can preserve, namely no `ids`, custom `item_func`, XML namespaces, XPath mode, root scalar payloads, or special `@` keys. At import time, the wrapper also verifies that an installed Rust backend rejects XML 1.0 forbidden characters; outdated or broken accelerators stay disabled so the Python security boundary cannot be bypassed.
 
 The backend adapter protocol exposes its diagnostic name as a read-only property, matching the frozen adapter implementations while still allowing selector code to inspect backend metadata.
 
@@ -68,6 +70,8 @@ Dependency floors and lockfiles keep known vulnerable packages out of runtime an
 
 Runtime dependencies are declared in `pyproject.toml` and mirrored by `uv.lock`; legacy requirements inputs remain pinned for tooling that still consumes requirements files. Security fixes should update both resolver paths so `uv audit` and requirements-based installs agree.
 
+Dependabot checks the root and documentation Python dependency manifests weekly, alongside the existing GitHub Actions monitoring, so stale security pins are surfaced automatically.
+
 ## Workflow supply-chain hardening
 
 GitHub Actions workflows run with read-only tokens by default and use full SHA pins so third-party action updates are explicit.