napalm_amt/record.py

@@ -107,7 +107,9 @@ def _landing_page(host: str, port: int, tls: bool, out: Path) -> None:
     """The page discovery fetches, without credentials, with its status line and headers."""
     url = f"{'https' if tls else 'http'}://{host}:{port}/"
     try:
-        response = requests.get(url, timeout=10, verify=False, allow_redirects=True)
+        # Recorded as discovery sees it: no credentials go out, and AMT's own certificate
+        # on 16993 is self-signed unless one was provisioned.
+        response = requests.get(url, timeout=10, verify=False, allow_redirects=True)  # nosec B501
     except requests.RequestException as exc:
         print(f"landing page: {exc}", file=sys.stderr)
         return

napalm_amt/wsman.py

@@ -13,14 +13,21 @@ The envelopes follow Intel's own go-wsman-messages, which every AMT since versio
 from __future__ import annotations
 
 import itertools
-import xml.etree.ElementTree as ET
 from collections.abc import Callable, Iterable, Mapping
-from typing import Any
+from typing import TYPE_CHECKING, Any
-from xml.sax.saxutils import escape, quoteattr
+
+# escape and quoteattr build the request envelopes; nothing here parses with SAX.
+from xml.sax.saxutils import escape, quoteattr  # nosec B406
 
 import requests
+from defusedxml import DefusedXmlException
+from defusedxml.ElementTree import ParseError, fromstring
 from requests.auth import HTTPDigestAuth
 
+if TYPE_CHECKING:
+    # For the annotations only; answers are parsed by defusedxml.
+    import xml.etree.ElementTree as ET  # nosec B405
+
 SOAP_NS = "http://www.w3.org/2003/05/soap-envelope"
 ADDRESSING_NS = "http://schemas.xmlsoap.org/ws/2004/08/addressing"
 WSMAN_NS = "http://schemas.dmtf.org/wbem/wsman/1/wsman.xsd"
@@ -267,13 +274,12 @@ class WsmanTransport:
         """The SOAP body of a response; a fault, a refusal or anything else raises."""
         if response.status_code == 401:
             raise WsmanAuthError(f"{self.url}: credentials refused (HTTP 401)")
-        body = None
+        # A DTD has no business in a WS-Management response; refusing one, in whatever
-        # A DTD has no business in a WS-Management response; refusing one keeps entity
+        # encoding, keeps entity expansion out of the parser.
-        # expansion out of the parser.
+        body: ET.Element | None
-        if b"<!DOCTYPE" not in response.content:
         try:
-                body = ET.fromstring(response.content).find(f"{{{SOAP_NS}}}Body")
+            body = fromstring(response.content, forbid_dtd=True).find(f"{{{SOAP_NS}}}Body")
-            except ET.ParseError:
+        except (ParseError, DefusedXmlException):
             body = None
         if body is not None:
             fault = body.find(f"{{{SOAP_NS}}}Fault")

pyproject.toml

@@ -14,6 +14,7 @@ dependencies = [
     # Not on PyPI: install from git before this package (see README).
     "napalm-device-types>=4.16",
     "requests>=2.31",
+    "defusedxml>=0.7",
 ]
 classifiers = [
     "License :: OSI Approved :: Apache Software License",

tests/test_wsman.py

@@ -214,6 +214,29 @@ def test_a_non_soap_error_raises():
             _transport(retries=0).get("AMT_GeneralSettings")
 
 
+_WITH_DTD = (
+    '<?xml version="1.0" encoding="{encoding}"?>'
+    '<!DOCTYPE Envelope [<!ENTITY a "aaaa">]>'
+    '<Envelope xmlns="http://www.w3.org/2003/05/soap-envelope"><Body>&a;</Body></Envelope>'
+)
+
+
+@pytest.mark.parametrize("encoding", ["utf-8", "utf-16"])
+def test_an_answer_with_a_dtd_is_refused(encoding):
+    """A DTD has no business in a WS-Management answer, in whatever encoding it comes."""
+    with responses.RequestsMock() as rsps:
+        rsps.add(responses.POST, URL, body=_WITH_DTD.format(encoding=encoding).encode(encoding))
+        with pytest.raises(WsmanError, match="no WS-Management answer"):
+            _transport(retries=0).get("AMT_GeneralSettings")
+
+
+def test_an_answer_that_is_no_xml_raises():
+    with responses.RequestsMock() as rsps:
+        rsps.add(responses.POST, URL, status=200, body=b"<Envelope><Body>")
+        with pytest.raises(WsmanError, match="HTTP 200"):
+            _transport(retries=0).get("AMT_GeneralSettings")
+
+
 def test_a_connection_error_is_retried_then_raised():
     with responses.RequestsMock() as rsps:
         rsps.add(responses.POST, URL, body=requests.ConnectionError("refused"))