utils.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

import base64
from collections.abc import Mapping
import hashlib
import re
from datetime import datetime, timezone
from typing import Any, Optional, Union
from urllib.parse import unquote
import xml.etree.ElementTree as ET
from xml.dom import minidom

from asn1crypto import csr
from asn1crypto import cms as a_cms, x509 as a_x509, core as a_core
from functools import lru_cache

from pathlib import Path
from cryptography.x509.oid import ExtensionOID

from cryptography import x509 as cx509
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.asymmetric import (
    padding,
    rsa,
    ec,
    ed25519,
    ed448,
)

import smtplib
import ssl
import time
import logging
import random
import struct
from contextlib import suppress
from threading import Lock
from email.message import EmailMessage
from email.utils import formataddr

try:
    from cryptography.hazmat.primitives.asymmetric import mldsa
except ImportError:  # cryptography versions without ML-DSA support
    mldsa = None
from cryptography.x509.oid import ObjectIdentifier as CObjectIdentifier
from utils_crt import is_certificate_revoked_by_crl

# Samba is optional: only the legacy search_user() uses it.
try:
    from samba.credentials import Credentials
    from samba.param import LoadParm
    from samba.samdb import SamDB
except ImportError:
    Credentials = LoadParm = SamDB = None

# pyasn1 for minimal CMC/PKIResponse structures
from pyasn1.type import univ, namedtype, namedval, char, useful, constraint
from pyasn1.codec.der.encoder import encode as der_encode

# DC discovery: DNS SRV + Netlogon LDAP Ping (TCP/389).
import dns.exception
import dns.name
import dns.resolver
from dns.rdtypes.IN.SRV import SRV
from ldap3 import BASE, NONE, Connection, Server
from ldap3.core.exceptions import LDAPException
from ldap3.utils.conv import escape_filter_chars
import gssapi
# -----------------------------------------------------------------------------
# Useful OIDs
# -----------------------------------------------------------------------------
OID_MS_CERT_TEMPLATE_NAME = "1.3.6.1.4.1.311.20.2"
OID_MS_CERT_TEMPLATE_INFO = "1.3.6.1.4.1.311.21.7"

# For PKIResponse (PKIResponse + CMCStatusInfo)
OID_ID_CCT_PKI_RESPONSE = "1.3.6.1.5.5.7.12.3"  # eContentType for PKIResponse
OID_ID_CMC_STATUS_INFO = "1.3.6.1.5.5.7.7.1"     # id-cmc-statusInfo
OID_MS_CMC_ADD_ATTRIBUTES = "1.3.6.1.4.1.311.10.10.1"
OID_MS_ISSUED_CERT_HASH = "1.3.6.1.4.1.311.21.17"


# -----------------------------------------------------------------------------
# ASN.1 structures (CMC/PKIData, NTDS CA Security, etc.)
# -----------------------------------------------------------------------------

class _TaggedCertificationRequest(a_cms.Sequence):
    _fields = [
        ("bodyPartID", a_cms.Integer),
        ("certificationRequest", csr.CertificationRequest),
    ]


class _CertReqMsg(a_core.Any):
    pass


class _ORM(a_core.Any):
    pass


class _TaggedRequest(a_cms.Choice):
    _alternatives = [
        ("tcr", _TaggedCertificationRequest, {"implicit": 0}),
        ("crm", _CertReqMsg, {"implicit": 1}),
        ("orm", _ORM, {"implicit": 2}),
    ]


class _TaggedRequests(a_cms.SequenceOf):
    _child_spec = _TaggedRequest


class _Controls(a_cms.SequenceOf):
    _child_spec = a_core.Any


class _TaggedContentInfos(a_cms.SequenceOf):
    _child_spec = a_core.Any


class _OtherMsgs(a_cms.SequenceOf):
    _child_spec = a_core.Any


class PKIData(a_cms.Sequence):
    _fields = [
        ("controlSequence", _Controls),
        ("reqSequence", _TaggedRequests),
        ("cmsSequence", _TaggedContentInfos),
        ("otherMsgSequence", _OtherMsgs),
    ]


class CertificateTemplate(a_core.Sequence):
    _fields = [
        ("template_id", a_core.ObjectIdentifier),
        ("major_version", a_core.Integer),
        ("minor_version", a_core.Integer),
    ]


class SMIMECapability(a_core.Sequence):
    _fields = [
        ("capability_id", a_core.ObjectIdentifier),
        ("parameters", a_core.Any, {"optional": True}),
    ]


class SMIMECapabilities(a_core.SequenceOf):
    _child_spec = SMIMECapability


class ExplicitOctetString0(a_core.Sequence):
    class_ = 2
    tag = 0
    _fields = [
        ('value', a_core.OctetString),
    ]

class NtdsOtherName(a_core.Sequence):
    class_ = 2
    tag = 0
    _fields = [
        ("type_id", a_core.ObjectIdentifier),    
        ("value", ExplicitOctetString0),    
    ]

class NtdsCASecurityExt(a_core.Sequence):
    _fields = [
        ("other_name", NtdsOtherName),
    ]


class PKIStatusInfo(a_core.Sequence):
    _fields = [
        ("status", a_core.Integer),
        ("status_string", a_core.SequenceOf, {"spec": a_core.UTF8String, "optional": True}),
        ("fail_info", a_core.BitString, {"optional": True}),
    ]


class CertOrEncCert(a_core.Choice):
    _alternatives = [
        ("certificate", a_x509.Certificate),
        ("encrypted_cert", a_core.Any),
    ]


class CertifiedKeyPair(a_core.Sequence):
    _fields = [
        ("cert_or_enc_cert", CertOrEncCert),
        ("private_key", a_core.Any, {"optional": True}),
        ("publication_info", a_core.Any, {"optional": True}),
    ]


class CertResponse(a_core.Sequence):
    _fields = [
        ("cert_req_id", a_core.Integer),
        ("status", PKIStatusInfo),
        ("certified_key_pair", CertifiedKeyPair, {"optional": True}),
        ("rsp_info", a_core.OctetString, {"optional": True}),
    ]


class CertRepMessage(a_core.Sequence):
    _fields = [
        ("ca_pubs", a_core.SequenceOf, {"spec": a_x509.Certificate, "optional": True}),
        ("response", a_core.SequenceOf, {"spec": CertResponse}),
    ]


class _PKIDataRelax(a_core.Sequence):
    _fields = [
        ("controlSequence", a_core.SequenceOf, {"spec": a_core.Any, "optional": True}),
        ("reqSequence", a_core.SequenceOf, {"spec": a_core.Any, "optional": True}),
        ("cmsSequence", a_core.SequenceOf, {"spec": a_core.Any, "optional": True}),
        ("otherMsgSequence", a_core.SequenceOf, {"spec": a_core.Any, "optional": True}),
    ]


class _TCRRelax(a_core.Sequence):
    _fields = [
        ("bodyPartID", a_core.Integer),
        ("certificationRequest", csr.CertificationRequest),
    ]


class _SeqOfCSR(a_core.SequenceOf):
    _child_spec = csr.CertificationRequest


class _MsCertTemplateInfo(a_core.Sequence):
    _fields = [
        ("templateID", a_core.ObjectIdentifier),
        ("majorVersion", a_core.Integer, {"optional": True}),
        ("minorVersion", a_core.Integer, {"optional": True}),
    ]


def _cert_has_template_oid(cert, oid: str) -> bool:
    try:
        ext_value = cert.extensions.get_extension_for_oid(
            CObjectIdentifier(OID_MS_CERT_TEMPLATE_INFO)
        ).value

        template_id = getattr(ext_value, "template_id", None)
        if template_id is not None:
            return getattr(template_id, "dotted_string", str(template_id)) == oid

        raw = getattr(ext_value, "value", None)
        if raw is None:
            return False

        return a_core.Sequence.load(bytes(raw))[0].native == oid
    except Exception:
        return False

# -----------------------------------------------------------------------------
# Small ASN.1 helpers
# -----------------------------------------------------------------------------

def _der_wrap_sequence(contents: bytes) -> bytes:
    """Wrap raw SEQUENCE contents with DER SEQUENCE header."""
    n = len(contents)
    if n < 128:
        return b"\x30" + bytes([n]) + contents
    lb = []
    x = n
    while x:
        lb.append(x & 0xFF)
        x >>= 8
    lb.reverse()
    return b"\x30" + bytes([0x80 | len(lb)]) + bytes(lb) + contents


# -----------------------------------------------------------------------------
# CSR parsing (extract CSR from CMC/PKIData and parse MS template)
# -----------------------------------------------------------------------------

def _parse_template_from_csr_bytes(csr_der: bytes) -> dict:
    """
    Return {"name","oid","major","minor"} if present in Microsoft Template
    (both Template Name and Template Info).
    """
    tpl = {"name": None, "oid": None, "major": None, "minor": None}

    csr_obj = csr.CertificationRequest.load(csr_der)
    cri = csr_obj["certification_request_info"]

    for attr in cri["attributes"]:
        # extensionRequest (1.2.840.113549.1.9.14)
        if attr["type"].dotted != "1.2.840.113549.1.9.14":
            continue

        for val in attr["values"]:
            exts = a_x509.Extensions.load(val.dump()) if not isinstance(val, a_x509.Extensions) else val
            for ext in exts:
                ext_oid = ext["extn_id"].dotted
                inner_der = ext["extn_value"].parsed.dump()

                if ext_oid == OID_MS_CERT_TEMPLATE_NAME:
                    # Try multiple string types; fallback on UTF-16-LE decode
                    name = None
                    for typ in (
                        a_core.BMPString,
                        a_core.UTF8String,
                        a_core.PrintableString,
                        a_core.TeletexString,
                    ):
                        try:
                            name = typ.load(inner_der).native
                            break
                        except Exception:
                            pass
                    if name is None:
                        try:
                            name = bytes(inner_der).decode("utf-16-le")
                        except Exception:
                            pass
                    tpl["name"] = name

                elif ext_oid == OID_MS_CERT_TEMPLATE_INFO:
                    info = _MsCertTemplateInfo.load(inner_der)
                    tpl["oid"] = info["templateID"].dotted
                    if info["majorVersion"].native is not None:
                        tpl["major"] = int(info["majorVersion"].native)
                    if info["minorVersion"].native is not None:
                        tpl["minor"] = int(info["minorVersion"].native)

    return tpl


MS_TPL_NAME_OID = "1.3.6.1.4.1.311.20.2"
MS_TPL_V2_OID = "1.3.6.1.4.1.311.21.7"
EXT_REQ_OID = "1.2.840.113549.1.9.14"  # extensionRequest

def _decode_ms_template_value(data: bytes) :
    """Try to decode the template value (often BMPString/UTF16)."""
    # 1) Generic ASN.1 attempt
    try:
        anyv = a_core.Asn1Value.load(data)
        if isinstance(anyv.native, str):
            return anyv.native
        if isinstance(anyv, a_core.OctetString):
            inner = anyv.native if isinstance(anyv.native, (bytes, bytearray)) else anyv.contents
            try:
                inner_any = a_core.Asn1Value.load(inner)
                if isinstance(inner_any.native, str):
                    return inner_any.native
            except Exception:
                pass
    except Exception:
        pass
    # 2) Fallback encodings
    for enc in ("utf-16-be", "utf-16-le", "utf-8", "latin1"):
        try:
            return data.decode(enc)
        except Exception:
            continue
    return None


def extract_ms_template_from_csr_der(csr_der: bytes):
    """
    Extract Microsoft template from a CSR (DER).
    Returns {"name": str|None, "oid": str|None}.
    Looks for both direct attributes and extensionRequest.
    """
    req = csr.CertificationRequest.load(csr_der)
    cri = req["certification_request_info"]
    attrs = cri["attributes"]

    tpl_name: Optional[str] = None
    tpl_oid: Optional[str] = None

    for attr in attrs:
        dotted = attr["type"].dotted

        # Direct attribute: template name
        if dotted == MS_TPL_NAME_OID and tpl_name is None:
            try:
                tpl_name = _decode_ms_template_value(attr["values"][0].contents)
            except Exception:
                pass

        # Direct attribute: template v2 OID
        if dotted == MS_TPL_V2_OID and tpl_oid is None:
            try:
                seq = a_core.Sequence.load(attr["values"][0].contents)
                tpl_oid = seq[0].native
            except Exception:
                pass

        # extensionRequest -> Extensions
        if dotted == EXT_REQ_OID:
            val0 = attr["values"][0]
            exts = getattr(val0, "parsed", val0)
            if isinstance(exts, a_x509.Extensions):
                for ext in exts:
                    ext_oid = ext["extn_id"].dotted
                    ev = ext["extn_value"]

                    if ext_oid == MS_TPL_V2_OID and tpl_oid is None:
                        parsed = getattr(ev, "parsed", None)
                        if parsed is not None:
                            try:
                                tpl_oid = parsed[0].native
                            except Exception:
                                pass
                        else:
                            data = ev.native if isinstance(ev.native, (bytes, bytearray)) else ev.contents
                            try:
                                seq = a_core.Sequence.load(data)
                                tpl_oid = seq[0].native
                            except Exception:
                                pass

                    if ext_oid == MS_TPL_NAME_OID and tpl_name is None:
                        data = ev.native if isinstance(ev.native, (bytes, bytearray)) else ev.contents
                        tpl_name = _decode_ms_template_value(data)

    return {"name": tpl_name, "oid": tpl_oid}


def _hash_for_name(name: str):
    algos = {
        "sha1": hashes.SHA1(),
        "sha224": hashes.SHA224(),
        "sha256": hashes.SHA256(),
        "sha384": hashes.SHA384(),
        "sha512": hashes.SHA512(),
    }
    algo = algos.get(name)
    if algo is None:
        raise ValueError(f"Unsupported digest algorithm: {name}")
    return algo


def _get_sd_certificates(sd) -> list:
    certs = []
    if sd["certificates"] is None:
        return certs

    for c in sd["certificates"]:
        if c.name == "certificate":
            certs.append(c.chosen)
    return certs


def _get_cert_ski(cert):
    """
    Return Subject Key Identifier raw bytes from an asn1crypto certificate,
    or None if missing.
    """
    try:
        exts = cert["tbs_certificate"]["extensions"]
    except Exception:
        return None

    for ext in exts:
        try:
            if ext["extn_id"].dotted == "2.5.29.14":
                parsed = ext["extn_value"].parsed
                return parsed.native
        except Exception:
            continue
    return None


def _find_signer_cert(sd, signer_info):
    certs = _get_sd_certificates(sd)
    sid = signer_info["sid"]

    if sid.name == "issuer_and_serial_number":
        iasn = sid.chosen
        issuer = iasn["issuer"]
        serial = iasn["serial_number"].native

        for cert in certs:
            try:
                if cert.issuer == issuer and cert.serial_number == serial:
                    return cert
            except Exception:
                continue

    elif sid.name == "subject_key_identifier":
        wanted_ski = sid.native
        for cert in certs:
            cert_ski = _get_cert_ski(cert)
            if cert_ski == wanted_ski:
                return cert

    return None


def _extract_message_digest_from_signed_attrs(signed_attrs):
    if signed_attrs is None:
        return None

    for attr in signed_attrs:
        try:
            if attr["type"].native == "message_digest":
                vals = attr["values"]
                if len(vals) > 0:
                    return vals[0].native
        except Exception:
            continue
    return None


def _extract_attr_value(signed_attrs, attr_name: str):
    """
    Return the first native value for a CMS signed attribute, or None.

    attr_name is the asn1crypto native name, for example:
      - "content_type"
      - "message_digest"

    Returns None when signed_attrs is None, the attribute is absent, or it has
    no values. Parsing errors on individual attributes are ignored so one bad
    attribute does not prevent checking the rest.
    """
    if signed_attrs is None:
        return None

    for attr in signed_attrs:
        try:
            if attr["type"].native == attr_name:
                vals = attr["values"]
                if len(vals) > 0:
                    return vals[0].native
        except Exception:
            continue
    return None


def _verify_cms_signature_with_cert(cert, signer_info, signed_bytes: bytes):
    cert_crypto = cx509.load_der_x509_certificate(cert.dump())
    pub = cert_crypto.public_key()

    signature = signer_info["signature"].native
    sig_algo = signer_info["signature_algorithm"]["algorithm"].native

    if isinstance(pub, rsa.RSAPublicKey):
        hash_name = signer_info["digest_algorithm"]["algorithm"].native
        hash_algo = _hash_for_name(hash_name)

        if sig_algo in ("rsassa_pkcs1v15", "sha1_rsa", "sha224_rsa", "sha256_rsa", "sha384_rsa", "sha512_rsa"):
            pub.verify(signature, signed_bytes, padding.PKCS1v15(), hash_algo)
            return

        if sig_algo == "rsassa_pss":
            # Optional PSS parameters may be present
            params = signer_info["signature_algorithm"]["parameters"]
            if params is not None and params.native:
                pss = params.native

                mgf = pss.get("mask_gen_algorithm", {})
                mgf_params = mgf.get("parameters", {})
                mgf_hash_name = mgf_params.get("algorithm", hash_name)
                mgf_hash_algo = _hash_for_name(mgf_hash_name)

                salt_len = pss.get("salt_length", hash_algo.digest_size)

                pub.verify(
                    signature,
                    signed_bytes,
                    padding.PSS(
                        mgf=padding.MGF1(mgf_hash_algo),
                        salt_length=salt_len,
                    ),
                    hash_algo,
                )
                return

            pub.verify(
                signature,
                signed_bytes,
                padding.PSS(
                    mgf=padding.MGF1(hash_algo),
                    salt_length=hash_algo.digest_size,
                ),
                hash_algo,
            )
            return

        raise ValueError(f"Unsupported RSA signature algorithm: {sig_algo}")

    if isinstance(pub, ec.EllipticCurvePublicKey):
        hash_name = signer_info["digest_algorithm"]["algorithm"].native
        hash_algo = _hash_for_name(hash_name)
        pub.verify(signature, signed_bytes, ec.ECDSA(hash_algo))
        return

    if isinstance(pub, ed25519.Ed25519PublicKey):
        pub.verify(signature, signed_bytes)
        return

    if isinstance(pub, ed448.Ed448PublicKey):
        pub.verify(signature, signed_bytes)
        return

    raise ValueError(f"Unsupported public key type: {type(pub).__name__}")


def _verify_signer_info(sd, signer_info):
    result = {
        "sid_type": signer_info["sid"].name,
        "subject": None,
        "issuer": None,
        "serial_number": None,
        "cert_der_b64": None,
        "signature_valid": False,
        "message_digest_valid": None,
        "errors": [],
    }

    cert = _find_signer_cert(sd, signer_info)
    if cert is None:
        result["errors"].append("Unable to match signerInfo.sid to a certificate")
        return result

    try:
        result["subject"] = cert.subject.human_friendly
    except Exception:
        pass
    try:
        result["issuer"] = cert.issuer.human_friendly
    except Exception:
        pass
    try:
        result["serial_number"] = cert.serial_number
    except Exception:
        pass
    try:
        result["cert_der_b64"] = base64.b64encode(cert.dump()).decode()
    except Exception:
        pass

    eci = sd["encap_content_info"]
    content = eci["content"].native if eci["content"] is not None else b""
    signed_attrs = signer_info["signed_attrs"]

    try:
        if signed_attrs is not None:
            ct_attr = _extract_attr_value(signed_attrs, "content_type")
            if ct_attr is None:
                result["errors"].append("signed_attrs present but no contentType attribute")
                return result

            expected_ct = eci["content_type"].native if eci["content_type"] is not None else None
            if ct_attr != expected_ct:
                result["errors"].append("contentType attribute does not match encapsulated content type")
                return result

            md_attr = _extract_message_digest_from_signed_attrs(signed_attrs)
            if md_attr is None:
                result["errors"].append("signed_attrs present but no messageDigest attribute")
                return result

            digest_name = signer_info["digest_algorithm"]["algorithm"].native
            h = hashlib.new(digest_name)
            h.update(content)
            computed_md = h.digest()

            result["message_digest_valid"] = (computed_md == md_attr)
            if not result["message_digest_valid"]:
                result["errors"].append("messageDigest attribute does not match encapsulated content")
                return result

            # In CMS, the signature covers the SET OF signedAttrs
            signed_bytes = _tbs_signed_attrs(signed_attrs)
        else:
            signed_bytes = content

        _verify_cms_signature_with_cert(cert, signer_info, signed_bytes)
        result["signature_valid"] = True
        return result

    except Exception as e:
        result["errors"].append(f"CMS signature verification failed: {e}")
        return result


def extract_and_verify_cmc_signers(sd):
    signers = []
    for signer_info in sd["signer_infos"]:
        signers.append(_verify_signer_info(sd, signer_info))
    return signers


def exct_csr_from_cmc(p7_der: bytes):
    """
    Try, in this order, to extract a CSR and its bodyPartID from a wrapped CMC
    (PKIData in SignedData), then fall back to simplePKIRequest or direct CSR.

    Returns:
        (csr_der, body_part_id, template_info_dict)

    template_info_dict is enriched with:
        - cmc_signatures
        - cmc_signature_valid
        - cmc_signature_errors
    """
    try:
        ci = a_cms.ContentInfo.load(p7_der)
        if ci["content_type"].native != "signed_data":
            raise ValueError("ContentInfo.content_type != signed_data")

        sd = ci["content"]
        eci = sd["encap_content_info"]

        if eci["content"] is None:
            raise ValueError("SignedData without encapsulated content (detached)")

        signers = extract_and_verify_cmc_signers(sd)
        pki_octets: bytes = eci["content"].native

        def _enrich_template_info(template_info: dict) -> dict:
            template_info = dict(template_info or {})
            template_info["cmc_signatures"] = signers
            template_info["cmc_signature_valid"] = any(
                s.get("signature_valid", False) for s in signers
            ) if signers else False
            if template_info["cmc_signature_valid"]:
                template_info["cmc_signature_errors"] = []
            else:
                template_info["cmc_signature_errors"] = [
                    err for s in signers for err in s.get("errors", [])
                ]
            return template_info

        # 1) PKIData -> look for TaggedCertificationRequest
        try:
            pkidata_relax = _PKIDataRelax.load(pki_octets)
            if pkidata_relax["reqSequence"] is not None:
                for any_item in pkidata_relax["reqSequence"]:
                    try:
                        wrapped = _der_wrap_sequence(any_item.contents)
                        tcr = _TCRRelax.load(wrapped)
                        csr_obj = tcr["certificationRequest"]
                        body_part_id = int(tcr["bodyPartID"].native)
                        csr_bytes = csr_obj.dump()
                        template_info = _enrich_template_info(
                            _parse_template_from_csr_bytes(csr_bytes)
                        )
                        return csr_bytes, body_part_id, template_info
                    except Exception:
                        continue
        except Exception:
            pass

        # 2) simplePKIRequest (SEQUENCE OF CSR)
        try:
            seq = _SeqOfCSR.load(pki_octets)
            if len(seq) >= 1:
                csr_bytes = seq[0].dump()
                body_part_id = 0
                template_info = _enrich_template_info(
                    _parse_template_from_csr_bytes(csr_bytes)
                )
                return csr_bytes, body_part_id, template_info
        except Exception:
            pass

        # 3) direct content = CSR
        try:
            csr_obj = csr.CertificationRequest.load(pki_octets)
            csr_bytes = csr_obj.dump()
            body_part_id = 0
            template_info = _enrich_template_info(
                _parse_template_from_csr_bytes(csr_bytes)
            )
            return csr_bytes, body_part_id, template_info
        except Exception:
            pass

        ct = eci["content_type"].dotted if eci["content_type"] is not None else "unknown"
        raise ValueError(
            "Unable to extract a CSR: no TCR, no simplePKIRequest, nor a direct CSR. "
            f"EncapContentInfo.content_type={ct}"
        )

    except Exception:
        # Ultimate fallback: some clients send just a "naked" CSR
        direct = csr.CertificationRequest.load(p7_der)
        csr_bytes = direct.dump()

        info = extract_ms_template_from_csr_der(csr_bytes)
        info["cmc_signatures"] = []
        info["cmc_signature_valid"] = False
        info["cmc_signature_errors"] = [
            "Request is a bare PKCS#10, not a CMS/CMC SignedData"
        ]
        return csr_bytes, 0, info


# -----------------------------------------------------------------------------
# CSR validation (explicit signature verification)
# -----------------------------------------------------------------------------

class CSRValidationError(ValueError):
    pass


def validate_csr(csr_obj: cx509.CertificateSigningRequest) -> None:
    """
    Validate a CSR:
      - verify its signature (RSA, ECDSA, Ed25519/Ed448, ML-DSA-44/65/87)
      - allow only supported public key types
      - enforce RSA >= 2048 bits with an odd exponent >= 65537
      - restrict EC keys to P-256, P-384, or P-521
    Raise CSRValidationError if invalid.
    """
    try:
        pub = csr_obj.public_key()
    except Exception as exc:
        raise CSRValidationError(f"Unable to read CSR public key: {exc}") from exc

    allowed_types = (
        rsa.RSAPublicKey,
        ec.EllipticCurvePublicKey,
        ed25519.Ed25519PublicKey,
        ed448.Ed448PublicKey,
    )
    if mldsa is not None:
        allowed_types += (
            mldsa.MLDSA44PublicKey,
            mldsa.MLDSA65PublicKey,
            mldsa.MLDSA87PublicKey,
        )

    if not isinstance(pub, allowed_types):
        raise CSRValidationError(
            f"Unsupported CSR public key type: {type(pub).__name__}"
        )

    # cryptography verifies the CSR signature with the appropriate algorithm,
    # including ML-DSA, without a separate signature_hash_algorithm or padding.
    try:
        valid_signature = csr_obj.is_signature_valid
    except Exception as exc:
        raise CSRValidationError(f"CSR signature verification failed: {exc}") from exc
    if not valid_signature:
        raise CSRValidationError("CSR signature verification failed: invalid signature")

    # Public key constraints
    if isinstance(pub, rsa.RSAPublicKey):
        key_size = pub.key_size
        if key_size < 2048:
            raise CSRValidationError(f"RSA key too small: {key_size} bits (min 2048)")
        try:
            e = pub.public_numbers().e
        except Exception as exc:
            raise CSRValidationError("Unable to read RSA public exponent") from exc
        if e < 65537 or e % 2 == 0:
            raise CSRValidationError(
                f"RSA public exponent not acceptable: {e} (must be odd >= 65537)"
            )
    elif isinstance(pub, ec.EllipticCurvePublicKey):
        curve = getattr(pub.curve, "name", str(pub.curve)).lower()
        allowed = {"secp256r1", "prime256v1", "secp384r1", "secp521r1"}
        if curve not in allowed:
            raise CSRValidationError(f"EC curve not allowed: {curve} (allowed: P-256/384/521)")
    # Ed25519/Ed448 and ML-DSA: no additional key constraints.


# -----------------------------------------------------------------------------
# BST / PKCS#7 (CES-like responses as in ADCS)
# -----------------------------------------------------------------------------

def _tbs_signed_attrs(attrs: a_cms.CMSAttributes) -> bytes:
    """
    Return the SET (tag 0x31) of the attributes for PKCS#7 signature.
    """
    der = attrs.dump()
    return (b"\x31" + der[1:]) if der and der[0] == 0xA0 else der



# -----------------------------------------------------------------------------
# PKIResponse (pending/denied) builders and CMS wrapper
# -----------------------------------------------------------------------------

class BodyPartID(univ.Integer):
    """BodyPartID ::= INTEGER (0..4294967295)"""
    subtypeSpec = univ.Integer.subtypeSpec + constraint.ValueRangeConstraint(0, 4294967295)


class BodyPartList(univ.SequenceOf):
    componentType = BodyPartID()


class CMCStatus(univ.Integer):
    namedValues = namedval.NamedValues(
        ('success', 0), ('failed', 2), ('pending', 3), ('noSupport', 4),
        ('confirmRequired', 5), ('popRequired', 6), ('partial', 7),
    )


class CMCFailInfo(univ.Integer):
    pass


class PendInfo(univ.Sequence):
    componentType = namedtype.NamedTypes(
        namedtype.NamedType('pendToken', univ.OctetString()),
        namedtype.NamedType('pendTime', useful.GeneralizedTime()),
    )


class OtherInfo(univ.Choice):
    componentType = namedtype.NamedTypes(
        namedtype.NamedType('failInfo', CMCFailInfo()),
        namedtype.NamedType('pendInfo', PendInfo()),
    )


class CMCStatusInfo(univ.Sequence):
    componentType = namedtype.NamedTypes(
        namedtype.NamedType('cMCStatus', CMCStatus()),
        namedtype.NamedType('bodyList', BodyPartList()),
        namedtype.OptionalNamedType('statusString', char.UTF8String()),
        namedtype.OptionalNamedType('otherInfo', OtherInfo()),
    )


class AttributeValue(univ.Any):
    pass


class AttributeValues(univ.SetOf):
    componentType = AttributeValue()


class CMCAttributeValue(univ.Any):
    pass


class CMCAttributeValues(univ.SetOf):
    componentType = CMCAttributeValue()


class CMCAttribute(univ.Sequence):
    componentType = namedtype.NamedTypes(
        namedtype.NamedType('type', univ.ObjectIdentifier()),
        namedtype.NamedType('values', CMCAttributeValues()),
    )


class CMCAttributes(univ.SetOf):
    componentType = CMCAttribute()


class BodyPartIDSequence(univ.SequenceOf):
    componentType = BodyPartID()


class CMCAddAttributes(univ.Sequence):
    componentType = namedtype.NamedTypes(
        namedtype.NamedType('pkiDataReference', BodyPartID()),
        namedtype.NamedType('certReferences', BodyPartIDSequence()),
        namedtype.NamedType('attributes', CMCAttributes()),
    )


class TaggedAttribute(univ.Sequence):
    componentType = namedtype.NamedTypes(
        namedtype.NamedType('bodyPartID', BodyPartID()),
        namedtype.NamedType('attrType', univ.ObjectIdentifier()),
        namedtype.NamedType('attrValues', AttributeValues()),
    )


class TaggedAttributeSeq(univ.SequenceOf):
    componentType = TaggedAttribute()


class TaggedContentInfo(univ.Sequence):
    pass  # unused


class TaggedContentInfoSeq(univ.SequenceOf):
    componentType = TaggedContentInfo()


class OtherMsg(univ.Sequence):
    pass


class OtherMsgSeq(univ.SequenceOf):
    componentType = OtherMsg()


class PKIResponse(univ.Sequence):
    componentType = namedtype.NamedTypes(
        namedtype.NamedType('controlSequence', TaggedAttributeSeq()),
        namedtype.NamedType('cmsSequence', TaggedContentInfoSeq()),
        namedtype.NamedType('otherMsgSequence', OtherMsgSeq()),
    )


def _build_pkiresponse_pending_der(
    request_id: int,
    status_text: str = "Pending",
    body_part_id: int = 1
) -> bytes:
    """Build a minimal PKIResponse with CMCStatusInfo(pending)."""
    csi = CMCStatusInfo()
    csi.setComponentByName('cMCStatus', 3)  # pending

    bl = BodyPartList()
    bl.append(BodyPartID(body_part_id))
    csi.setComponentByName('bodyList', bl)

    if status_text:
        csi.setComponentByName('statusString', char.UTF8String(status_text))

    pend = PendInfo()
    pend.setComponentByName('pendToken', univ.OctetString(str(request_id).encode('ascii')))
    pend.setComponentByName('pendTime', useful.GeneralizedTime(datetime.now(timezone.utc).strftime('%Y%m%d%H%M%SZ')))
    oi = OtherInfo()
    oi.setComponentByName('pendInfo', pend)
    csi.setComponentByName('otherInfo', oi)

    ta = TaggedAttribute()
    ta.setComponentByName('bodyPartID', BodyPartID(body_part_id))
    ta.setComponentByName('attrType', univ.ObjectIdentifier(OID_ID_CMC_STATUS_INFO))
    avs = AttributeValues()
    avs.append(AttributeValue(der_encode(csi)))
    ta.setComponentByName('attrValues', avs)

    controls = TaggedAttributeSeq()
    controls.append(ta)

    resp = PKIResponse()
    resp.setComponentByName('controlSequence', controls)
    resp.setComponentByName('cmsSequence', TaggedContentInfoSeq())
    resp.setComponentByName('otherMsgSequence', OtherMsgSeq())

    return der_encode(resp)


def _build_pkiresponse_denied_der(
    status_text: str = "Denied by administrator",
    body_part_id: int = 1
) -> bytes:
    """Build a minimal PKIResponse with CMCStatusInfo(failed)."""
    csi = CMCStatusInfo()
    csi.setComponentByName('cMCStatus', 2)  # failed

    bl = BodyPartList()
    bl.append(BodyPartID(body_part_id))
    csi.setComponentByName('bodyList', bl)

    if status_text:
        csi.setComponentByName('statusString', char.UTF8String(status_text))

    ta = TaggedAttribute()
    ta.setComponentByName('bodyPartID', BodyPartID(body_part_id))
    ta.setComponentByName('attrType', univ.ObjectIdentifier(OID_ID_CMC_STATUS_INFO))
    avs = AttributeValues()
    avs.append(AttributeValue(der_encode(csi)))
    ta.setComponentByName('attrValues', avs)

    controls = TaggedAttributeSeq()
    controls.append(ta)

    resp = PKIResponse()
    resp.setComponentByName('controlSequence', controls)
    resp.setComponentByName('cmsSequence', TaggedContentInfoSeq())
    resp.setComponentByName('otherMsgSequence', OtherMsgSeq())

    return der_encode(resp)


def _build_pkiresponse_issued_der(
    cert_der: bytes,
    *,
    status_text: str = "Issued",
    cert_req_id: int = 1,
) -> bytes:
    """
    Build a PKIResponse(success) matching ADCS CMC full PKI response format.

    Per MS-WCCE 3.2.1.4.2.1.4.8.2, the PKIResponse controlSequence MUST contain:
      - bodyPartID=1, id-cmc-statusInfo, CMCStatusInfo(success)
      - bodyPartID=2, szOID_CMC_ADD_ATTRIBUTES with szOID_ISSUED_CERT_HASH = SHA1(issued cert)
    cmsSequence and otherMsgSequence are unused.
    """
    csi = CMCStatusInfo()
    csi.setComponentByName('cMCStatus', 0)  # success

    bl = BodyPartList()
    bl.append(BodyPartID(cert_req_id))
    csi.setComponentByName('bodyList', bl)

    if status_text:
        csi.setComponentByName('statusString', char.UTF8String(status_text))

    ta_status = TaggedAttribute()
    ta_status.setComponentByName('bodyPartID', BodyPartID(1))
    ta_status.setComponentByName('attrType', univ.ObjectIdentifier(OID_ID_CMC_STATUS_INFO))
    ta_status_values = AttributeValues()
    ta_status_values.append(AttributeValue(der_encode(csi)))
    ta_status.setComponentByName('attrValues', ta_status_values)

    cert_hash = hashlib.sha1(cert_der).digest()
    issued_hash_attr = CMCAttribute()
    issued_hash_attr.setComponentByName('type', univ.ObjectIdentifier(OID_MS_ISSUED_CERT_HASH))
    issued_hash_values = CMCAttributeValues()
    issued_hash_values.append(CMCAttributeValue(der_encode(univ.OctetString(cert_hash))))
    issued_hash_attr.setComponentByName('values', issued_hash_values)

    attrs = CMCAttributes()
    attrs.append(issued_hash_attr)

    cert_refs = BodyPartIDSequence()
    cert_refs.append(BodyPartID(cert_req_id))

    cmc_add = CMCAddAttributes()
    cmc_add.setComponentByName('pkiDataReference', BodyPartID(0))
    cmc_add.setComponentByName('certReferences', cert_refs)
    cmc_add.setComponentByName('attributes', attrs)

    ta_attrs = TaggedAttribute()
    ta_attrs.setComponentByName('bodyPartID', BodyPartID(2))
    ta_attrs.setComponentByName('attrType', univ.ObjectIdentifier(OID_MS_CMC_ADD_ATTRIBUTES))
    ta_attrs_values = AttributeValues()
    ta_attrs_values.append(AttributeValue(der_encode(cmc_add)))
    ta_attrs.setComponentByName('attrValues', ta_attrs_values)

    controls = TaggedAttributeSeq()
    controls.append(ta_status)
    controls.append(ta_attrs)

    resp = PKIResponse()
    resp.setComponentByName('controlSequence', controls)
    resp.setComponentByName('cmsSequence', TaggedContentInfoSeq())
    resp.setComponentByName('otherMsgSequence', OtherMsgSeq())
    return der_encode(resp)


def build_adcs_bst_pkiresponse_issued(
    child_der: bytes,
    ca_der: bytes,
    ca_key,
    cert_req_id: int = 1,
    *,
    status_text: str = "Issued",
) -> bytes:
    """Build the final CMC full PKI response used by ADCS for an issued cert."""
    leaf_cert = a_x509.Certificate.load(child_der)
    signer_cert = a_x509.Certificate.load(ca_der)

    pkiresp_der = _build_pkiresponse_issued_der(
        cert_der=child_der,
        status_text=status_text,
        cert_req_id=cert_req_id,
    )

    encap_content_info = a_cms.EncapsulatedContentInfo(
        {
            "content_type": a_cms.ContentType(OID_ID_CCT_PKI_RESPONSE),
            "content": a_cms.ParsableOctetString(pkiresp_der),
        }
    )

    signed_attrs = a_cms.CMSAttributes(
        [
            a_cms.CMSAttribute({"type": "1.2.840.113549.1.9.3", "values": [a_cms.ContentType(OID_ID_CCT_PKI_RESPONSE)]}),
            a_cms.CMSAttribute({"type": "1.2.840.113549.1.9.4", "values": [hashlib.sha256(pkiresp_der).digest()]}),
        ]
    )
    to_be_signed = _tbs_signed_attrs(signed_attrs)
    if ca_key :
        signer_info = [a_cms.SignerInfo(
            {
                "version": "v1",
                "sid": a_cms.SignerIdentifier(
                    {
                        "issuer_and_serial_number": a_cms.IssuerAndSerialNumber(
                            {"issuer": signer_cert.issuer, "serial_number": signer_cert.serial_number}
                        )
                    }
                ),
                "digest_algorithm": a_cms.DigestAlgorithm({"algorithm": "sha256"}),
                "signed_attrs": signed_attrs,
                "signature_algorithm": _signature_algo_for_ca_key(ca_key) if ca_key else a_cms.SignedDigestAlgorithm({"algorithm": "sha256_rsa"}),
                "signature": _sign_tbs_with_ca_key(ca_key, to_be_signed),
            }
        )]
    else:
        signer_info = []
    
    sd = a_cms.SignedData(
        {
            "version": 3,
            "digest_algorithms": [a_cms.DigestAlgorithm({"algorithm": "sha256"})],
            "encap_content_info": encap_content_info,
            "certificates": [leaf_cert, signer_cert],
            "signer_infos": signer_info,
        }
    )
    ci = a_cms.ContentInfo({"content_type": "signed_data", "content": sd})
    return ci.dump()


def build_adcs_bst_pkiresponse(
    ca_der: bytes,
    ca_key,
    request_id: int,
    status: str = "pending",
    status_text: str = "Pending",
    body_part_id: int = 1
) -> bytes:
    """
    Build a CMS SignedData containing a PKIResponse (Windows/ADCS-ready).
    - eContentType = 1.3.6.1.5.5.7.12.3
    - SignedAttributes = content-type(12.3) + message-digest(SHA256(PKIResponse))
    Parameters:
      - status: "pending" or "denied" (aliases: "failed", "refused")
    """
    signer_cert = a_x509.Certificate.load(ca_der)

    st = (status or "").strip().lower()
    if st in ("pending", "pend", "p"):
        pkiresp_der = _build_pkiresponse_pending_der(
            request_id=request_id,
            status_text=status_text,
            body_part_id=body_part_id,
        )
    elif st in ("denied", "failed", "refused", "fail"):
        pkiresp_der = _build_pkiresponse_denied_der(
            status_text=status_text,
            body_part_id=body_part_id,
        )
    else:
        raise ValueError(f"Unsupported PKIResponse status: {status!r} (expected: 'pending' or 'denied')")

    encap_content_info = a_cms.EncapsulatedContentInfo(
        {
            "content_type": a_cms.ContentType(OID_ID_CCT_PKI_RESPONSE),
            "content": a_cms.ParsableOctetString(pkiresp_der),
        }
    )

    signed_attrs = a_cms.CMSAttributes(
        [
            a_cms.CMSAttribute({"type": "1.2.840.113549.1.9.3", "values": [a_cms.ContentType(OID_ID_CCT_PKI_RESPONSE)]}),
            a_cms.CMSAttribute({"type": "1.2.840.113549.1.9.4", "values": [hashlib.sha256(pkiresp_der).digest()]}),
        ]
    )
    to_be_signed = _tbs_signed_attrs(signed_attrs)

    signer_info = a_cms.SignerInfo(
        {
            "version": "v1",
            "sid": a_cms.SignerIdentifier(
                {
                    "issuer_and_serial_number": a_cms.IssuerAndSerialNumber(
                        {"issuer": signer_cert.issuer, "serial_number": signer_cert.serial_number}
                    )
                }
            ),
            "digest_algorithm": a_cms.DigestAlgorithm({"algorithm": "sha256"}),
            "signed_attrs": signed_attrs,
            "signature_algorithm": _signature_algo_for_ca_key(ca_key) if ca_key else a_cms.SignedDigestAlgorithm({"algorithm": "sha256_rsa"}),
            "signature": b"",
        }
    )
    if ca_key:
        sig = _sign_tbs_with_ca_key(ca_key, to_be_signed)
    else:
        sig = b''

    signer_info["signature"] = sig

    sd = a_cms.SignedData(
        {
            "version": 3,
            "digest_algorithms": [a_cms.DigestAlgorithm({"algorithm": "sha256"})],
            "encap_content_info": encap_content_info,
            "certificates": [signer_cert],
            "signer_infos": [signer_info],
        }
    )
    ci = a_cms.ContentInfo({"content_type": "signed_data", "content": sd})
    return ci.dump()


# -----------------------------------------------------------------------------
# Utility encoding (Base64 helpers)
# -----------------------------------------------------------------------------

def format_b64_for_soap(data: bytes) -> str:
    """Base64-encode bytes for inclusion into SOAP text nodes."""
    return base64.b64encode(data).decode("ascii")


def _wrap_b64(data: bytes, line_len: int = 64, with_crlf: bool = True) -> str:
    """
    Base64-encode bytes and optionally wrap lines to a fixed length.
    If with_crlf is True, lines are joined using CRLF (\r\n).
    """
    s = base64.b64encode(data).decode("ascii")
    if not line_len:
        return s
    lines = [s[i:i + line_len] for i in range(0, len(s), line_len)]
    sep = "\r\n" if with_crlf else "\n"
    return sep.join(lines)


# -----------------------------------------------------------------------------
# WS-Trust (Enrollment) response builder
# -----------------------------------------------------------------------------

def build_ws_trust_response(
    pkcs7_der: bytes,
    relates_to: str,
    request_id: int,
    ces_uri: str,
    *,
    status: str = "pending",  # "pending" | "denied"/"failed"/"fail"/"refused"
    disposition_message: str = "Pending",
    reason_text: str = "Denied by administrator",
    error_code: int = -2146877420,
    invalid_request: bool = True,
    lang: str = "fr-FR",
    token_type: str = "http://docs.oasis-open.org/wss/2004/01/oasis-200401-wss-x509-token-profile-1.0#X509v3",
    activity_correlation_id: str = "3fd81603-cf85-48d8-945d-990e2a7a5673",
    activity_id: str = "00000000-0000-0000-0000-000000000000",
    wrap_b64_lines: bool = True,
):
    """
    Build a WS-Trust response:
      - status="pending"  -> RSTRC (success), returns (xml_bytes, 200)
      - status in {"denied","failed","fail","refused"} -> SOAP Fault (CertificateEnrollmentWSDetailFault), returns (xml_bytes, 500)
    """
    import base64
    import xml.etree.ElementTree as ET
    from xml.etree.ElementTree import QName

    NS = {
        "s":  "http://www.w3.org/2003/05/soap-envelope",
        "a":  "http://www.w3.org/2005/08/addressing",
        "wst":"http://docs.oasis-open.org/ws-sx/ws-trust/200512",
        "wss":"http://docs.oasis-open.org/wss/2004/01/oasis-200401-wss-wssecurity-secext-1.0.xsd",
        "msdiag": "http://schemas.microsoft.com/2004/09/ServiceModel/Diagnostics",
        "msenroll": "http://schemas.microsoft.com/windows/pki/2009/01/enrollment",
        "xsd": "http://www.w3.org/2001/XMLSchema",
        "xsi": "http://www.w3.org/2001/XMLSchema-instance",
    }
    for p, uri in NS.items():
        ET.register_namespace(p, uri)

    def _wrap_b64_crlf(data: bytes, line_len: int = 64) -> str:
        s = base64.b64encode(data).decode("ascii")
        if not line_len:
            return s
        return "\r\n".join(s[i:i+line_len] for i in range(0, len(s), line_len))

    st = (status or "").strip().lower()
    is_pending = st in ("pending", "pend", "p")
    is_failed = st in ("denied", "failed", "fail", "refused")

    if is_pending:
        # ----- RSTRC (HTTP 200) -----
        senv = ET.Element(QName(NS["s"], "Envelope"))

        header = ET.SubElement(senv, QName(NS["s"], "Header"))
        a_action = ET.SubElement(header, QName(NS["a"], "Action"), {QName(NS["s"], "mustUnderstand"): "1"})
        a_action.text = "http://schemas.microsoft.com/windows/pki/2009/01/enrollment/RSTRC/wstep"
        a_rel = ET.SubElement(header, QName(NS["a"], "RelatesTo")); a_rel.text = relates_to
        activity = ET.SubElement(header, QName(NS["msdiag"], "ActivityId"), {"CorrelationId": activity_correlation_id})
        activity.text = activity_id

        body = ET.SubElement(senv, QName(NS["s"], "Body"))
        rstrc = ET.SubElement(body, QName(NS["wst"], "RequestSecurityTokenResponseCollection"))
        rstr = ET.SubElement(rstrc, QName(NS["wst"], "RequestSecurityTokenResponse"))

        tokentype = ET.SubElement(rstr, QName(NS["wst"], "TokenType"))
        tokentype.text = token_type

        disp = ET.SubElement(rstr, QName(NS["msenroll"], "DispositionMessage"), {"{http://www.w3.org/XML/1998/namespace}lang": lang})
        disp.text = disposition_message

        bst = ET.SubElement(
            rstr,
            QName(NS["wss"], "BinarySecurityToken"),
            {"ValueType": f"{NS['wss']}#PKCS7", "EncodingType": f"{NS['wss']}#base64binary"},
        )
        bst.text = _wrap_b64_crlf(pkcs7_der) if wrap_b64_lines else base64.b64encode(pkcs7_der).decode("ascii")

        req_tok = ET.SubElement(rstr, QName(NS["wst"], "RequestedSecurityToken"))
        str_el = ET.SubElement(req_tok, QName(NS["wss"], "SecurityTokenReference"))
        ET.SubElement(str_el, QName(NS["wss"], "Reference"), {"URI": ces_uri})

        req_id = ET.SubElement(rstr, QName(NS["msenroll"], "RequestID")); req_id.text = str(request_id)

        return ET.tostring(senv, encoding="utf-8", xml_declaration=True, method="xml"), 200

    elif is_failed:
        # ----- SOAP Fault (HTTP 500): CertificateEnrollmentWSDetailFault -----
        senv = ET.Element(QName(NS["s"], "Envelope"))

        header = ET.SubElement(senv, QName(NS["s"], "Header"))
        a_action = ET.SubElement(header, QName(NS["a"], "Action"), {QName(NS["s"], "mustUnderstand"): "1"})
        a_action.text = "http://schemas.microsoft.com/windows/pki/2009/01/enrollment/RequestSecurityTokenCertificateEnrollmentWSDetailFault"
        a_rel = ET.SubElement(header, QName(NS["a"], "RelatesTo")); a_rel.text = relates_to
        activity = ET.SubElement(header, QName(NS["msdiag"], "ActivityId"), {"CorrelationId": activity_correlation_id})
        activity.text = activity_id

        body = ET.SubElement(senv, QName(NS["s"], "Body"))
        fault = ET.SubElement(body, QName(NS["s"], "Fault"))

        code = ET.SubElement(fault, QName(NS["s"], "Code"))
        ET.SubElement(code, QName(NS["s"], "Value")).text = "s:Receiver"

        reason = ET.SubElement(fault, QName(NS["s"], "Reason"))
        ET.SubElement(reason, QName(NS["s"], "Text"), {"{http://www.w3.org/XML/1998/namespace}lang": lang}).text = reason_text

        detail = ET.SubElement(fault, QName(NS["s"], "Detail"))
        cert_detail = ET.SubElement(
            detail,
            QName(NS["msenroll"], "CertificateEnrollmentWSDetail"),
            {"xmlns:xsd": NS["xsd"], "xmlns:xsi": NS["xsi"]},
        )

        b64 = _wrap_b64_crlf(pkcs7_der) if wrap_b64_lines else base64.b64encode(pkcs7_der).decode("ascii")
        ET.SubElement(cert_detail, QName(NS["msenroll"], "BinaryResponse")).text = b64
        ET.SubElement(cert_detail, QName(NS["msenroll"], "ErrorCode")).text = str(int(error_code))
        ET.SubElement(cert_detail, QName(NS["msenroll"], "InvalidRequest")).text = "true" if invalid_request else "false"
        ET.SubElement(cert_detail, QName(NS["msenroll"], "RequestID")).text = str(request_id)

        return ET.tostring(senv, encoding="utf-8", xml_declaration=True, method="xml"), 500

    else:
        raise ValueError(f"Unsupported status '{status}': use 'pending' or 'denied'")


# -----------------------------------------------------------------------------
# Active Directory: DNS/LDAP DC discovery and Samba/LDAP user lookup
# -----------------------------------------------------------------------------

# Helpers for ldap3 Connection.response entries (case-insensitive AD attributes).
def _attribute(entry: Any, section: str, name: str) -> Any:
    """Return the first value of an attribute, or None if it is absent."""
    if not isinstance(entry, Mapping) or not isinstance(name, str):
        return None
    attrs = entry.get(section)
    if not isinstance(attrs, Mapping):
        return None
    wanted = name.casefold()
    for key, value in attrs.items():
        if isinstance(key, str) and key.casefold() == wanted:
            if isinstance(value, (tuple, list)):
                return value[0] if value else None
            return value
    return None


def ldap_text(entry: Any, name: str) -> Optional[str]:
    """Read one textual LDAP attribute, or None if unavailable/invalid."""
    value = _attribute(entry, "attributes", name)
    if value is None:
        value = _attribute(entry, "raw_attributes", name)

    if isinstance(value, str):
        return value if value and "\x00" not in value else None
    if isinstance(value, (bytes, bytearray, memoryview)):
        try:
            decoded = bytes(value).decode("utf-8")
        except UnicodeDecodeError:
            return None
        return decoded if decoded and "\x00" not in decoded else None
    return None


def ldap_raw(entry: Any, name: str) -> Optional[bytes]:
    """Read the bytes of one LDAP attribute, e.g. AD's objectGUID."""
    value = _attribute(entry, "raw_attributes", name)
    if value is None:
        value = _attribute(entry, "attributes", name)
    if isinstance(value, (bytes, bytearray, memoryview)):
        return bytes(value)
    return None


def _binary_sid(value: bytes) -> Optional[str]:
    """Decode an MS-DTYP SID: revision, authority, LE subauthorities."""
    if len(value) < 12 or value[0] != 1:
        return None
    count = value[1]
    if not 1 <= count <= 15 or len(value) != 8 + 4 * count:
        return None
    authority = int.from_bytes(value[2:8], "big")
    parts = [str(int.from_bytes(value[8 + i * 4:12 + i * 4], "little"))
             for i in range(count)]
    return "S-1-{}-{}".format(authority, "-".join(parts))


def _text_sid(value: str) -> Optional[str]:
    parts = value.split("-")
    if len(parts) < 4 or len(parts) > 18 or parts[:2] != ["S", "1"]:
        return None
    if any(not part or not part.isascii() or not part.isdecimal()
           for part in parts[2:]):
        return None
    # Bound string lengths before converting untrusted LDAP values to ints.
    if len(parts[2]) > 15 or any(len(part) > 10 for part in parts[3:]):
        return None
    nums = [int(part) for part in parts[2:]]
    if nums[0] >= 1 << 48 or any(number >= 1 << 32 for number in nums[1:]):
        return None
    return "S-1-" + "-".join(str(number) for number in nums)


def ldap_sid(entry: Any) -> Optional[str]:
    """Return a canonical objectSid such as S-1-5-21-..., or None.

    Prefer the original binary bytes to a transformed LDAP3 value. If binary
    data is present but invalid, refuse it instead of trusting a second copy.
    """
    raw = _attribute(entry, "raw_attributes", "objectSid")
    if raw is not None:
        if isinstance(raw, (bytes, bytearray, memoryview)):
            return _binary_sid(bytes(raw))
        return None

    parsed = _attribute(entry, "attributes", "objectSid")
    if isinstance(parsed, str):
        return _text_sid(parsed)
    if isinstance(parsed, (bytes, bytearray, memoryview)):
        return _binary_sid(bytes(parsed))
    return None


BAD_LDAP_CHARS_RE = re.compile(r'[\x00()*\\]')


log = logging.getLogger(__name__)
_dc_cache: dict[str, tuple[str, float]] = {}
_cache_lock = Lock()


def _parse_netlogon(data: bytes, realm: str) -> tuple[str, str]:
    """Decode NETLOGON_SAM_LOGON_RESPONSE_EX: (FQDN, client site)."""
    if len(data) < 40:
        raise ValueError("Reponse Netlogon trop courte")

    opcode, reserved, flags = struct.unpack_from("<HHI", data)
    # LOGON_SAM_LOGON_RESPONSE_EX and DS_LDAP_FLAG.
    if opcode != 23 or reserved != 0 or not (flags & 0x08):
        raise ValueError("DC indisponible ou reponse Netlogon inattendue")

    offset = 24  # Opcode + reserved + flags + 16-byte GUID.
    names = []
    for _ in range(8):
        name, consumed = dns.name.from_wire(data, offset)
        offset += consumed  # from_wire returns the number of bytes consumed, not an offset.
        names.append(b".".join(name.labels[:-1]).decode("utf-8"))

    # NtVer=4 requires neither an IP address nor an additional site field.
    if len(data) != offset + 8:
        raise ValueError("Longueur Netlogon inattendue")
    version, token1, token2 = struct.unpack_from("<IHH", data, offset)
    if not (version & 0x04) or (token1, token2) != (0xFFFF, 0xFFFF):
        raise ValueError("Terminaison Netlogon invalide")
    if names[1].lower() != realm or not names[2]:
        raise ValueError("Domaine inattendu ou FQDN absent")

    # forest, domain, host, netbios_domain, netbios_host, user, dc_site, client_site
    return names[2], names[7]


def _ldap_ping(host: str, port: int, realm: str, timeout: int) -> tuple[str, str]:
    server = Server(host, port=port, get_info=NONE, connect_timeout=timeout)
    conn = Connection(
        server,
        check_names=False,
        raise_exceptions=True,
        auto_referrals=False,
        receive_timeout=timeout,
    )
    try:
        # No bind: perform only an anonymous search on the rootDSE.
        conn.open(read_server_info=False)
        ok = conn.search(
            search_base="",
            search_filter=(
                f"(&(DnsDomain={escape_filter_chars(realm)})"
                r"(NtVer=\04\00\00\00))"
            ),
            search_scope=BASE,
            attributes=["Netlogon"],
            time_limit=timeout,
        )
        if ok:
            for entry in conn.response:
                for name, values in entry.get("raw_attributes", {}).items():
                    if name.lower() == "netlogon" and values:
                        return _parse_netlogon(values[0], realm)
        raise ValueError(f"{host}: aucune reponse Netlogon exploitable")
    finally:
        with suppress(LDAPException, OSError):
            conn.unbind()


def _srv_records(realm: str, site: Optional[str], timeout: int) -> list[SRV]:
    # Treat the site name as a single DNS label, even if it contains a dot.
    labels = [b"_ldap", b"_tcp"]
    if site:
        labels += [site.encode("utf-8"), b"_sites"]
    labels += [b"dc", b"_msdcs"]
    query = dns.name.Name(labels) + dns.name.from_text(realm)
    answer = dns.resolver.resolve(query, "SRV", lifetime=timeout, search=False)
    records = [r for r in answer if r.target != dns.name.root]

    # Ascending SRV priority, then weighted selection (RFC 2782).
    ordered = []
    for priority in sorted({r.priority for r in records}):
        group = [r for r in records if r.priority == priority]
        random.shuffle(group)
        group.sort(key=lambda r: r.weight != 0)
        while group:
            draw = random.randint(0, sum(r.weight for r in group))
            cumulative = 0
            for i, record in enumerate(group):
                cumulative += record.weight
                if cumulative >= draw:
                    ordered.append(group.pop(i))
                    break
    return ordered


def _first_dc(records: list[SRV], realm: str, timeout: int) -> tuple[str, str]:
    last_error = None
    for record in records:
        host = record.target.to_text(omit_final_dot=True)
        try:
            return _ldap_ping(host, record.port, realm, timeout)
        except (LDAPException, OSError, ValueError, dns.exception.DNSException) as exc:
            last_error = exc
            log.debug("DC ignore: %s: %s", host, exc)
    raise RuntimeError(f"Aucun DC LDAP utilisable pour {realm}") from last_error


def get_dc_fqdn(
    realm: str,
    dc_cache_ttl_seconds: float = 3600,
    *,
    timeout: int = 3,
) -> str:
    """Return a DC, preferably in the client site. Cache for 1 hour by default."""
    realm = realm.strip().rstrip(".").lower()
    if not realm:
        raise ValueError("realm ne doit pas etre vide")
    if dc_cache_ttl_seconds < 0:
        raise ValueError("Le TTL doit etre positif ou nul")
    if not isinstance(timeout, int) or timeout <= 0:
        raise ValueError("timeout doit etre un entier strictement positif")

    with _cache_lock:
        cached = _dc_cache.get(realm)
        if dc_cache_ttl_seconds > 0 and cached and time.monotonic() < cached[1]:
            return cached[0]

    # 1. Discover a domain DC and query it for the client site.
    initial_dc, client_site = _first_dc(_srv_records(realm, None, timeout), realm, timeout)
    chosen_dc = initial_dc

    # 2. Prefer a DC in the client site; otherwise keep the initial valid DC.
    if client_site:
        try:
            chosen_dc, _ = _first_dc(
                _srv_records(realm, client_site, timeout), realm, timeout
            )
        except (dns.exception.DNSException, RuntimeError) as exc:
            log.debug("Repli sur %s: %s", initial_dc, exc)

    # Application-level TTL, as in the original function, not the DNS TTL.
    with _cache_lock:
        if dc_cache_ttl_seconds > 0:
            _dc_cache[realm] = (chosen_dc, time.monotonic() + dc_cache_ttl_seconds)
        else:
            _dc_cache.pop(realm, None)
    return chosen_dc


def search_user(userauth: str,ldap_filter='',dc_fqdn=None,basedn=None,password=None,bind_user=None,dc_cache_ttl_seconds=300):
    """
    Resolve the SAM/LDAP entry for the Kerberos user 'user@REALM'.
    Returns (SamDB, entry) if found.
    """

    if Credentials is None or LoadParm is None or SamDB is None:
        raise ImportError("Samba is required by search_user(); use search_user_ldap3() without Samba")

    username = userauth.split("@", 1)[0].strip()

    if BAD_LDAP_CHARS_RE.search(username):
        raise ValueError("invalid username")

    lp = LoadParm()
    lp.load_default()

    creds = Credentials()
    creds.guess(lp)
    if bind_user:
        if not password:
            raise ValueError("password is required when bind_user is set")
        bind_username = bind_user.split("@", 1)[0].strip()
        if BAD_LDAP_CHARS_RE.search(bind_username):
            raise ValueError("invalid bind username")
        creds.set_username(bind_username)
        creds.set_password(password)
    elif not password:
        creds.set_kerberos_state(True)
        creds.set_machine_account(lp)
    else:
        creds.set_username(username)
        creds.set_password(password)

    realm = lp.get("realm")
    if not dc_fqdn:
        dc_fqdn = get_dc_fqdn(realm,dc_cache_ttl_seconds=dc_cache_ttl_seconds)

    ldap_url = f"ldap://{dc_fqdn}"

    samdbr = SamDB(url=ldap_url, credentials=creds, lp=lp)
    if not basedn:
        if "@" in userauth:
            basedn = "DC=" + userauth.split("@")[1].replace(".", ",DC=")
        else:
            basedn = samdbr.get_default_basedn()
        
    ldap_filter = "(&(samAccountName=%s)%s)" % (username,ldap_filter)

    res = samdbr.search(
        base=basedn,
        scope=2,
        expression=ldap_filter
    )

    for entry in res:
        return samdbr, entry


def search_user_ldap3(
    userauth: str,
    ldap_filter: str = "",
    dc_fqdn: Optional[str] = None,
    basedn: Optional[str] = None,
    password: Optional[str] = None,
    bind_user: Optional[str] = None,
    dc_cache_ttl_seconds: float = 300,
    *,
    realm: Optional[str] = None,
    keytab: str = "/etc/krb5.keytab",
    attributes: Optional[Union[list[str], tuple[str, ...]]] = None,
    timeout: int = 10,
) -> Optional[dict]:
    """LDAP3 + Kerberos SASL, with GSSAPI encryption when available.

    No TLS; lazy import keeps the legacy Samba path independent of gssapi.
    """
    from ldap3_auth import find_user

    return find_user(
        userauth=userauth,
        ldap_filter=ldap_filter,
        dc_fqdn=dc_fqdn,
        basedn=basedn,
        password=password,
        bind_user=bind_user,
        dc_cache_ttl_seconds=dc_cache_ttl_seconds,
        realm=realm,
        keytab=keytab,
        attributes=attributes,
        timeout=timeout,
        get_dc_fqdn=get_dc_fqdn,
    )


# -----------------------------------------------------------------------------
# Static extensions (EKU/KU/AppPolicies/TemplateInfo) applied to the cert
# -----------------------------------------------------------------------------

def _apply_static_extensions(builder: cx509.CertificateBuilder, template: dict) -> cx509.CertificateBuilder:
    """
    Re-read static extensions already materialized in the template (field __der)
    and re-apply them to x509.CertificateBuilder, typing EKU/KU when possible.
    Extensions marked dynamic (__der=None) are ignored here.
    """
    EKU_OID = "2.5.29.37"
    KU_OID = "2.5.29.15"

    src = template.get("__all_required_extensions") or template.get("required_extensions") or []
    for ext in src:
        der = ext.get("__der")
        if der is None:
            continue  # dynamic (e.g., NTDS) -> handled elsewhere
        oid = ext.get("oid")
        critical = bool(ext.get("critical", False))

        if oid == EKU_OID:
            try:
                eku_asn1 = a_x509.ExtKeyUsageSyntax.load(der)
                oids = [cx509.ObjectIdentifier(str(x)) for x in eku_asn1]
                eku = cx509.ExtendedKeyUsage(oids)
                builder = builder.add_extension(eku, critical=critical)
            except Exception:
                builder = builder.add_extension(
                    cx509.UnrecognizedExtension(CObjectIdentifier(oid), der),
                    critical=critical,
                )

        elif oid == KU_OID:
            try:
                bits = a_core.BitString.load(der).native
                bits += [False] * (9 - len(bits))
                ku = cx509.KeyUsage(
                    digital_signature=bits[0],
                    content_commitment=bits[1],
                    key_encipherment=bits[2],
                    data_encipherment=bits[3],
                    key_agreement=bits[4],
                    key_cert_sign=bits[5],
                    crl_sign=bits[6],
                    encipher_only=bits[7],
                    decipher_only=bits[8],
                )
                builder = builder.add_extension(ku, critical=critical)
            except Exception:
                builder = builder.add_extension(
                    cx509.UnrecognizedExtension(CObjectIdentifier(oid), der),
                    critical=True,
                )
        else:
            # AppPolicies (1.3.6.1.4.1.311.21.10), TemplateInfo (1.3.6.1.4.1.311.21.7), etc.
            builder = builder.add_extension(
                cx509.UnrecognizedExtension(CObjectIdentifier(oid), der),
                critical=critical,
            )

    return builder


# -----------------------------------------------------------------------------
# Enrollment Policy (GetPolicies) XML helpers and response
# -----------------------------------------------------------------------------

NS_EP = {
    "s":   "http://www.w3.org/2003/05/soap-envelope",
    "a":   "http://www.w3.org/2005/08/addressing",
    "xsi": "http://www.w3.org/2001/XMLSchema-instance",
    "xsd": "http://www.w3.org/2001/XMLSchema",
    "ep":  "http://schemas.microsoft.com/windows/pki/2009/01/enrollmentpolicy",
    "diag":"http://schemas.microsoft.com/2004/09/ServiceModel/Diagnostics",
}
# Register once
ET.register_namespace('s',   NS_EP["s"])
ET.register_namespace('a',   NS_EP["a"])
ET.register_namespace('xsi', NS_EP["xsi"])
ET.register_namespace('xsd', NS_EP["xsd"])
ET.register_namespace('ep',  NS_EP["ep"])
ET.register_namespace('',    NS_EP["diag"])  # optional: limit prefix creation on ActivityId

_TRUE = "true"
_FALSE = "false"

def tbool(v: bool) -> str:
    return _TRUE if bool(v) else _FALSE

def text(elem, value):
    elem.text = "" if value is None else str(value)
    return elem

def set_xsi_nil(elem, is_nil: bool):
    elem.set(f"{{{NS_EP['xsi']}}}nil", _TRUE if is_nil else _FALSE)
    return elem

def prettify(xml_bytes: bytes) -> str:
    parsed = minidom.parseString(xml_bytes)
    return parsed.toprettyxml(indent="  ", encoding="utf-8").decode("utf-8")

def build_get_policies_response(
    uuid_request: str,
    uuid_random: str,
    hosturl: str,
    policyid: str,
    policyfriendlyname: str,
    next_update_hours: int,
    cas: list,
    templates: list,
    oids: list,
) -> str:
    # --- helpers for dynamic OID refs ---
    def _to_int_or_none(x):
        try:
            return int(str(x))
        except Exception:
            return None

    def _next_free_refid(oids_list: list) -> int:
        used = set()
        for oo in oids_list:
            v = oo.get("__refid") or oo.get("oIDReferenceID") or oo.get("refid")
            iv = _to_int_or_none(v)
            if iv is not None:
                used.add(iv)
        return (max(used) + 1) if used else 1

    def _find_oid_refid(oids_list: list, value: str, group: int, default_name: str):
        for oo in oids_list:
            if (
                str(oo.get("value")) == value
                and str(oo.get("group")) == str(group)
                and str(oo.get("default_name")) == default_name
            ):
                return oo.get("__refid")
        return None

    def _ensure_oid(oids_list: list, value: str, group: int, default_name: str, refid: Optional[int] = None) -> int:
        existing = _find_oid_refid(oids_list, value, group, default_name)
        if existing is not None:
            return int(existing)

        if refid is None:
            refid = _next_free_refid(oids_list)

        oids_list.append({
            "value": value,
            "group": group,
            "__refid": refid,
            "default_name": default_name,
        })
        return refid

    # --- ECDSA/ECDH curves mapping by minimalKeyLength ---
    # Note: ECC uses 256/384/521. If you pass 512 we map to 521.
    ECDSA_CURVES = {
        256: {"value": "1.2.840.10045.3.1.7", "group": 3, "default_name": "ECDSA_P256"},
        384: {"value": "1.3.132.0.34",       "group": 3, "default_name": "ECDSA_P384"},
        521: {"value": "1.3.132.0.35",       "group": 3, "default_name": "ECDSA_P521"},
    }
    ECDH_CURVES = {
        256: {"value": "1.2.840.10045.3.1.7", "group": 3, "default_name": "ECDH_P256"},
        384: {"value": "1.3.132.0.34",       "group": 3, "default_name": "ECDH_P384"},
        521: {"value": "1.3.132.0.35",       "group": 3, "default_name": "ECDH_P521"},
    }

    # --- Supported algorithms (RSA/DSA/ML-DSA are single OID entries) ---
    ALGO_OIDS = {
        "rsa": {"value": "1.2.840.113549.1.1.1", "group": 3, "default_name": "RSA"},
        "dsa": {"value": "1.2.840.10040.4.1",    "group": 3, "default_name": "DSA"},

        # ML-DSA / FIPS 204 / RFC 9882
        "ml-dsa-44": {"value": "2.16.840.1.101.3.4.3.17", "group": 3, "default_name": "ML-DSA-44"},
        "ml-dsa-65": {"value": "2.16.840.1.101.3.4.3.18", "group": 3, "default_name": "ML-DSA-65"},
        "ml-dsa-87": {"value": "2.16.840.1.101.3.4.3.19", "group": 3, "default_name": "ML-DSA-87"},

        # ecc handled separately via curves + minimalKeyLength (+ keySpec for ECDH)
    }

    # --- Hash algorithms OIDs (for hashAlgorithmOIDReference) ---
    # group=1 matches what ADCS emits for hash OIDs in XCEP responses
    HASH_OIDS = {
        "SHA1":   {"value": "1.3.14.3.2.26",            "group": 1, "default_name": "SHA1"},
        "SHA256": {"value": "2.16.840.1.101.3.4.2.1",   "group": 1, "default_name": "SHA256"},
        "SHA384": {"value": "2.16.840.1.101.3.4.2.2",   "group": 1, "default_name": "SHA384"},
        "SHA512": {"value": "2.16.840.1.101.3.4.2.3",   "group": 1, "default_name": "SHA512"},
    }

    # map algo_key -> allocated refid (reuse same refid across templates)
    # (algo_key = "rsa", "dsa", "ml-dsa-65", or "ecdsa:256"/"ecdh:256"/...)
    algo_refids: dict[str, int] = {}

    # <s:Envelope>
    env = ET.Element(ET.QName(NS_EP['s'], 'Envelope'))

    # <s:Header>
    hdr = ET.SubElement(env, ET.QName(NS_EP['s'], 'Header'))
    action = ET.SubElement(hdr, ET.QName(NS_EP['a'], 'Action'), {ET.QName(NS_EP['s'], 'mustUnderstand'): "1"})
    action.text = "http://schemas.microsoft.com/windows/pki/2009/01/enrollmentpolicy/IPolicy/GetPoliciesResponse"

    relates = ET.SubElement(hdr, ET.QName(NS_EP['a'], 'RelatesTo'))
    relates.text = f"urn:uuid:{uuid_request}"

    # ActivityId (Diagnostics ns) — keep it in diag ns like ADCS
    act = ET.SubElement(
        hdr,
        "ActivityId",
        {"CorrelationId": str(uuid_random), "xmlns": NS_EP["diag"]},
    )
    act.text = "00000000-0000-0000-0000-000000000000"

    # <s:Body>
    body = ET.SubElement(env, ET.QName(NS_EP['s'], 'Body'))
    body.set('xmlns:xsi', NS_EP['xsi'])
    body.set('xmlns:xsd', NS_EP['xsd'])

    # <ep:GetPoliciesResponse>
    gpr = ET.SubElement(body, ET.QName(NS_EP['ep'], 'GetPoliciesResponse'))

    # <response>
    response = ET.SubElement(gpr, ET.QName(NS_EP['ep'], 'response'))

    text(ET.SubElement(response, ET.QName(NS_EP['ep'], 'policyID')), policyid)
    pf = ET.SubElement(response, ET.QName(NS_EP['ep'], 'policyFriendlyName'))
    text(pf, policyfriendlyname) 
    text(ET.SubElement(response, ET.QName(NS_EP['ep'], 'nextUpdateHours')), next_update_hours)

    pol_not_changed = ET.SubElement(response, ET.QName(NS_EP['ep'], 'policiesNotChanged'))
    set_xsi_nil(pol_not_changed, True)

    policies = ET.SubElement(response, ET.QName(NS_EP['ep'], 'policies'))

    # --- templates -> policies/policy ---
    ca_allowed_refids = set()
    for t in templates:
        if t['permissions']['enroll']:
            for refid in t.get("__ca_refids") or []:
                ca_allowed_refids.add(refid)
        policy = ET.SubElement(policies, ET.QName(NS_EP['ep'], 'policy'))
        text(ET.SubElement(policy, ET.QName(NS_EP['ep'], 'policyOIDReference')), t["__policy_oid_reference"])

        cAs = ET.SubElement(policy, ET.QName(NS_EP['ep'], 'cAs'))
        ca_refids = t.get("__ca_refids") or []
        if ca_refids:
            for refid in ca_refids:
                text(ET.SubElement(cAs, ET.QName(NS_EP['ep'], 'cAReference')), refid)
        else:
            text(ET.SubElement(cAs, ET.QName(NS_EP['ep'], 'cAReference')), cas[0]["__refid"])

        attrs = ET.SubElement(policy, ET.QName(NS_EP['ep'], 'attributes'))
        text(ET.SubElement(attrs, ET.QName(NS_EP['ep'], 'commonName')), t["common_name"])
        text(ET.SubElement(attrs, ET.QName(NS_EP['ep'], 'policySchema')), t["policy_schema"])

        cert_valid = ET.SubElement(attrs, ET.QName(NS_EP['ep'], 'certificateValidity'))
        text(ET.SubElement(cert_valid, ET.QName(NS_EP['ep'], 'validityPeriodSeconds')), t["validity"]["validity_seconds"])
        text(ET.SubElement(cert_valid, ET.QName(NS_EP['ep'], 'renewalPeriodSeconds')), t["validity"]["renewal_seconds"])

        perm = ET.SubElement(attrs, ET.QName(NS_EP['ep'], 'permission'))
        text(ET.SubElement(perm, ET.QName(NS_EP['ep'], 'enroll')), tbool(t["permissions"]["enroll"]))
        text(ET.SubElement(perm, ET.QName(NS_EP['ep'], 'autoEnroll')), tbool(t["permissions"]["auto_enroll"]))

        pka = ET.SubElement(attrs, ET.QName(NS_EP['ep'], 'privateKeyAttributes'))
        text(ET.SubElement(pka, ET.QName(NS_EP['ep'], 'minimalKeyLength')), t["private_key_attributes"]["minimal_key_length"])
        text(ET.SubElement(pka, ET.QName(NS_EP['ep'], 'keySpec')), t["private_key_attributes"]["key_spec"])

        kup = ET.SubElement(pka, ET.QName(NS_EP['ep'], 'keyUsageProperty'))
        set_xsi_nil(kup, True)

        pk_perms = ET.SubElement(pka, ET.QName(NS_EP['ep'], 'permissions'))
        set_xsi_nil(pk_perms, True)

        # --- algorithmOIDReference (RSA/ECDSA/ECDH/DSA/ML-DSA) ---
        algo = (
            t.get("private_key_attributes", {})
            .get("algorithm", "")
            .strip()
            .lower()
            .replace("_", "-")
        )
        algo_aliases = {
            "mldsa44": "ml-dsa-44",
            "mldsa65": "ml-dsa-65",
            "mldsa87": "ml-dsa-87",
        }
        algo = algo_aliases.get(algo.replace("-", ""), algo)

        # Decide EC signature vs key-exchange:
        # - If algo explicitly "ecdh" => ECDH
        # - If algo is "ecdsa" => ECDSA (but if keySpec indicates key exchange, switch to ECDH to mimic ADCS template behavior)
        # - Otherwise keep existing behavior
        key_spec = _to_int_or_none(t.get("private_key_attributes", {}).get("key_spec"))
        ec_mode = None  # "ecdsa" or "ecdh"
        if algo == "ecdh":
            ec_mode = "ecdh"
        elif algo == "ecdsa":
            # ADCS templates: keySpec=1 typically means key exchange, keySpec=2 means signature
            if key_spec == 1:
                ec_mode = "ecdh"
            else:
                ec_mode = "ecdsa"

        if ec_mode in ("ecdsa", "ecdh"):
            bits = _to_int_or_none(t.get("private_key_attributes", {}).get("minimal_key_length"))
            if bits == 512:
                bits = 521  # tolerate "512" for P-521
            if bits not in (256, 384, 521):
                raise ValueError(f"Unsupported EC minimalKeyLength={bits}. Expected 256/384/521 (or 512->521).")

            meta = (ECDH_CURVES if ec_mode == "ecdh" else ECDSA_CURVES)[bits]
            algo_key = f"{ec_mode}:{bits}"

            if algo_key not in algo_refids:
                existing = _find_oid_refid(oids, meta["value"], meta["group"], meta["default_name"])
                if existing is not None:
                    algo_refids[algo_key] = int(existing)
                else:
                    algo_refids[algo_key] = _ensure_oid(
                        oids,
                        meta["value"],
                        meta["group"],
                        meta["default_name"],
                        refid=_next_free_refid(oids),
                    )

            refid = algo_refids[algo_key]
            alg_el = ET.SubElement(pka, ET.QName(NS_EP["ep"], "algorithmOIDReference"))
            alg_el.text = str(refid)

        elif algo in ALGO_OIDS:
            meta = ALGO_OIDS[algo]
            algo_key = algo

            if algo_key not in algo_refids:
                existing = _find_oid_refid(oids, meta["value"], meta["group"], meta["default_name"])
                if existing is not None:
                    algo_refids[algo_key] = int(existing)
                else:
                    algo_refids[algo_key] = _ensure_oid(
                        oids,
                        meta["value"],
                        meta["group"],
                        meta["default_name"],
                        refid=_next_free_refid(oids),
                    )

            refid = algo_refids[algo_key]
            alg_el = ET.SubElement(pka, ET.QName(NS_EP["ep"], "algorithmOIDReference"))
            alg_el.text = str(refid)

        else:
            # default / existing behavior
            alg_el = ET.SubElement(pka, ET.QName(NS_EP['ep'], 'algorithmOIDReference'))
            alg_ref = t["private_key_attributes"].get("algorithm_oid_reference")
            set_xsi_nil(alg_el, False if alg_ref else True)
            if alg_ref:
                alg_el.text = str(alg_ref)

        cp = ET.SubElement(pka, ET.QName(NS_EP['ep'], 'cryptoProviders'))
        for prov in t["private_key_attributes"].get("crypto_providers", []):
            text(ET.SubElement(cp, ET.QName(NS_EP['ep'], 'provider')), prov)

        rev = ET.SubElement(attrs, ET.QName(NS_EP['ep'], 'revision'))
        text(ET.SubElement(rev, ET.QName(NS_EP['ep'], 'majorRevision')), t["revision"]["major"])
        text(ET.SubElement(rev, ET.QName(NS_EP['ep'], 'minorRevision')), t["revision"]["minor"])

        sup = ET.SubElement(attrs, ET.QName(NS_EP['ep'], 'supersededPolicies'))
        set_xsi_nil(sup, True)

        text(ET.SubElement(attrs, ET.QName(NS_EP['ep'], 'privateKeyFlags')), t["flags"]["private_key_flags"])
        text(ET.SubElement(attrs, ET.QName(NS_EP['ep'], 'subjectNameFlags')), t["flags"]["subject_name_flags"])
        text(ET.SubElement(attrs, ET.QName(NS_EP['ep'], 'enrollmentFlags')), t["flags"]["enrollment_flags"])
        text(ET.SubElement(attrs, ET.QName(NS_EP['ep'], 'generalFlags')), t["flags"]["general_flags"])

        # --- hashAlgorithmOIDReference (optional) ---
        # priority:
        #   1) t["hash_algorithm"] (ex: "SHA256") -> allocate OID + set ref
        #   2) fallback: t["flags"]["hash_algorithm_oid_reference"] (existing behavior)
        desired_hash = (t.get("hash_algorithm") or t.get("flags", {}).get("hash_algorithm") or "").strip()
        hash_ref = None

        if desired_hash:
            key = desired_hash.upper()
            if key not in HASH_OIDS:
                raise ValueError(f"Unsupported hash_algorithm='{desired_hash}'. Expected one of {', '.join(HASH_OIDS.keys())}")

            meta = HASH_OIDS[key]
            existing = _find_oid_refid(oids, meta["value"], meta["group"], meta["default_name"])
            if existing is not None:
                hash_ref = int(existing)
            else:
                hash_ref = _ensure_oid(
                    oids,
                    meta["value"],
                    meta["group"],
                    meta["default_name"],
                    refid=_next_free_refid(oids),
                )

            # keep it also in flags for traceability/debug
            t.setdefault("flags", {})["hash_algorithm_oid_reference"] = hash_ref
        else:
            hash_ref = t.get("flags", {}).get("hash_algorithm_oid_reference")

        hash_alg = ET.SubElement(attrs, ET.QName(NS_EP['ep'], 'hashAlgorithmOIDReference'))
        if not hash_ref:
            set_xsi_nil(hash_alg, False if hash_ref else True)
        if hash_ref:
            hash_alg.text = str(hash_ref)

        ra_req = ET.SubElement(attrs, ET.QName(NS_EP['ep'], 'rARequirements'))
        set_xsi_nil(ra_req, True)

        kaa = ET.SubElement(attrs, ET.QName(NS_EP['ep'], 'keyArchivalAttributes'))
        set_xsi_nil(kaa, True)

        exts = ET.SubElement(attrs, ET.QName(NS_EP['ep'], 'extensions'))
        for ext in t.get("required_extensions", []):
            e = ET.SubElement(exts, ET.QName(NS_EP['ep'], 'extension'))
            text(ET.SubElement(e, ET.QName(NS_EP['ep'], 'oIDReference')), ext["__oid_reference"])
            text(ET.SubElement(e, ET.QName(NS_EP['ep'], 'critical')), tbool(ext["critical"]))
            text(ET.SubElement(e, ET.QName(NS_EP['ep'], 'value')), ext["value_b64"])

    ep_cas = ET.SubElement(gpr, ET.QName(NS_EP['ep'], 'cAs'))

    AUTH_MAP = {
        "anonymous": 1,
        "kerberos": 2,
        "username_password": 4,
        "x509": 8,
    }

    DEFAULT_AUTH_METHODS = [
        {"method": "kerberos", "renewal_only": False},
#        {"method": "username_password", "renewal_only": False},
#        {"method": "x509", "renewal_only": False},
    ]
    
    priority = 1 
    
    for ca in cas:
        ca_el = ET.SubElement(ep_cas, ET.QName(NS_EP['ep'], 'cA'))
        uris = ET.SubElement(ca_el, ET.QName(NS_EP['ep'], 'uris'))
    
        auth_entries = ca.get("auth_methods") or DEFAULT_AUTH_METHODS
        seen = set()
    
        for entry in auth_entries:
            method = (entry.get("method") or "").strip().lower()
            renewal_only = bool(entry.get("renewal_only", False))
    
            if method not in AUTH_MAP:
                raise ValueError(f"Unknown auth method '{method}' for CA '{ca.get('id', '?')}'")

            auth_type = AUTH_MAP[method]

            if auth_type in seen:
                continue
            seen.add(auth_type)

            cauri = ET.SubElement(uris, ET.QName(NS_EP['ep'], 'cAURI'))
            text(ET.SubElement(cauri, ET.QName(NS_EP['ep'], 'clientAuthentication')), auth_type)
            text(ET.SubElement(cauri, ET.QName(NS_EP['ep'], 'uri')), f"{hosturl}{ca['__ces_path']}")
            text(ET.SubElement(cauri, ET.QName(NS_EP['ep'], 'priority')), priority)
            text(ET.SubElement(cauri, ET.QName(NS_EP['ep'], 'renewalOnly')), "true" if renewal_only else "false")
            priority += 1

        text(ET.SubElement(ca_el, ET.QName(NS_EP['ep'], 'certificate')), ca["__certificate_b64"])
        text(ET.SubElement(ca_el, ET.QName(NS_EP['ep'], 'enrollPermission')), tbool(ca['__refid'] in ca_allowed_refids))
        text(ET.SubElement(ca_el, ET.QName(NS_EP['ep'], 'cAReferenceID')), ca["__refid"])


    # --- oIDs ---
    ep_oids = ET.SubElement(gpr, ET.QName(NS_EP['ep'], 'oIDs'))
    for o in oids:
        oe = ET.SubElement(ep_oids, ET.QName(NS_EP['ep'], 'oID'))
        text(ET.SubElement(oe, ET.QName(NS_EP['ep'], 'value')), o["value"])
        text(ET.SubElement(oe, ET.QName(NS_EP['ep'], 'group')), o["group"])
        text(ET.SubElement(oe, ET.QName(NS_EP['ep'], 'oIDReferenceID')), o["__refid"])
        text(ET.SubElement(oe, ET.QName(NS_EP['ep'], 'defaultName')), o["default_name"])

    raw = ET.tostring(env, encoding="utf-8", xml_declaration=True)
    return prettify(raw)


# -----------------------------------------------------------------------------
# CES response (Issued) builder
# -----------------------------------------------------------------------------

NS_CES = {
    "s":    "http://www.w3.org/2003/05/soap-envelope",
    "a":    "http://www.w3.org/2005/08/addressing",
    "diag": "http://schemas.microsoft.com/2004/09/ServiceModel/Diagnostics",
    "wst":  "http://docs.oasis-open.org/ws-sx/ws-trust/200512",
    "wsse": "http://docs.oasis-open.org/wss/2004/01/oasis-200401-wss-wssecurity-secext-1.0.xsd",
    "ep":   "http://schemas.microsoft.com/windows/pki/2009/01/enrollment",
}
ET.register_namespace('s',    NS_CES['s'])
ET.register_namespace('a',    NS_CES['a'])
ET.register_namespace('wst',  NS_CES['wst'])
ET.register_namespace('wsse', NS_CES['wsse'])
ET.register_namespace('ep',   NS_CES['ep'])

def _txt(elem, value):
    elem.text = "" if value is None else str(value)
    return elem

def _prettify(xml_bytes: bytes) -> str:
    return minidom.parseString(xml_bytes).toprettyxml(indent="  ", encoding="utf-8").decode("utf-8")

def build_ces_response(uuid_request: str, uuid_random: str, p7b_der: str, leaf_der: str, body_part_id: str) -> str:
    env = ET.Element(ET.QName(NS_CES['s'], 'Envelope'))

    hdr = ET.SubElement(env, ET.QName(NS_CES['s'], 'Header'))
    action = ET.SubElement(hdr, ET.QName(NS_CES['a'], 'Action'), {ET.QName(NS_CES['s'], 'mustUnderstand'): "1"})
    _txt(action, "http://schemas.microsoft.com/windows/pki/2009/01/enrollment/RSTRC/wstep")

    relates = ET.SubElement(hdr, ET.QName(NS_CES['a'], 'RelatesTo'))
    _txt(relates, f"urn:uuid:{uuid_request}")

    act = ET.SubElement(hdr, ET.QName(NS_CES['diag'], 'ActivityId'), {"CorrelationId": str(uuid_random)})
    _txt(act, "00000000-0000-0000-0000-000000000000")

    body = ET.SubElement(env, ET.QName(NS_CES['s'], 'Body'))

    rstrc = ET.SubElement(body, ET.QName(NS_CES['wst'], 'RequestSecurityTokenResponseCollection'))
    rstr = ET.SubElement(rstrc, ET.QName(NS_CES['wst'], 'RequestSecurityTokenResponse'))

    tok_type = ET.SubElement(rstr, ET.QName(NS_CES['wst'], 'TokenType'))
    _txt(tok_type, "http://docs.oasis-open.org/wss/2004/01/oasis-200401-wss-x509-token-profile-1.0#X509v3")

    disp = ET.SubElement(rstr, ET.QName(NS_CES['ep'], 'DispositionMessage'))
    disp.set('xml:lang', 'en-US')
    _txt(disp, "Issued")

    bst_p7b = ET.SubElement(
        rstr, ET.QName(NS_CES['wsse'], 'BinarySecurityToken'),
        {
            "ValueType":    "http://docs.oasis-open.org/wss/2004/01/oasis-200401-wss-wssecurity-secext-1.0.xsd#PKCS7",
            "EncodingType": "http://docs.oasis-open.org/wss/2004/01/oasis-200401-wss-wssecurity-secext-1.0.xsd#base64binary",
        }
    )
    _txt(bst_p7b, p7b_der)

    req_tok = ET.SubElement(rstr, ET.QName(NS_CES['wst'], 'RequestedSecurityToken'))

    bst_leaf = ET.SubElement(
        req_tok, ET.QName(NS_CES['wsse'], 'BinarySecurityToken'),
        {
            "ValueType":    "http://docs.oasis-open.org/wss/2004/01/oasis-200401-wss-x509-token-profile-1.0#X509v3",
            "EncodingType": "http://docs.oasis-open.org/wss/2004/01/oasis-200401-wss-wssecurity-secext-1.0.xsd#base64binary",
        }
    )
    _txt(bst_leaf, leaf_der)

    req_id = ET.SubElement(rstr, ET.QName(NS_CES['ep'], 'RequestID'))
    _txt(req_id, body_part_id)

    raw = ET.tostring(env, encoding="utf-8", xml_declaration=True)
    return _prettify(raw)


def build_ket_response(uuid_request: str, uuid_random: str, ket_cert_der: str) -> str:
    """
    Build CES/WSTEP WS-Trust RSTR/KETFinal response.

    - uuid_request: the UUID from the request MessageID (without "urn:uuid:" prefix is fine; we add it)
    - uuid_random: correlation id for diag:ActivityId
    - ket_cert_der: base64 of the DER-encoded X.509 certificate to return as the KeyExchangeToken (BinarySecurityToken)
    """
    env = ET.Element(ET.QName(NS_CES['s'], 'Envelope'))

    hdr = ET.SubElement(env, ET.QName(NS_CES['s'], 'Header'))

    action = ET.SubElement(hdr, ET.QName(NS_CES['a'], 'Action'), {ET.QName(NS_CES['s'], 'mustUnderstand'): "1"})
    _txt(action, "http://docs.oasis-open.org/ws-sx/ws-trust/200512/RSTR/KETFinal")

    relates = ET.SubElement(hdr, ET.QName(NS_CES['a'], 'RelatesTo'))
    _txt(relates, f"urn:uuid:{uuid_request}")

    act = ET.SubElement(hdr, ET.QName(NS_CES['diag'], 'ActivityId'), {"CorrelationId": str(uuid_random)})
    _txt(act, "00000000-0000-0000-0000-000000000000")

    body = ET.SubElement(env, ET.QName(NS_CES['s'], 'Body'))

    rstrc = ET.SubElement(body, ET.QName(NS_CES['wst'], 'RequestSecurityTokenResponseCollection'))
    rstr = ET.SubElement(rstrc, ET.QName(NS_CES['wst'], 'RequestSecurityTokenResponse'))

    tok_type = ET.SubElement(rstr, ET.QName(NS_CES['wst'], 'TokenType'))
    _txt(tok_type, "http://docs.oasis-open.org/wss/2004/01/oasis-200401-wss-x509-token-profile-1.0#X509v3")

    req_tok = ET.SubElement(rstr, ET.QName(NS_CES['wst'], 'RequestedSecurityToken'))
    ket = ET.SubElement(req_tok, ET.QName(NS_CES['wst'], 'KeyExchangeToken'))

    bst_ket = ET.SubElement(
        ket, ET.QName(NS_CES['wsse'], 'BinarySecurityToken'),
        {
            "ValueType":    "http://docs.oasis-open.org/wss/2004/01/oasis-200401-wss-x509-token-profile-1.0#X509v3",
            "EncodingType": "http://docs.oasis-open.org/wss/2004/01/oasis-200401-wss-wssecurity-secext-1.0.xsd#base64binary",
        }
    )
    _txt(bst_ket, ket_cert_der)

    raw = ET.tostring(env, encoding="utf-8", xml_declaration=True)
    return _prettify(raw)



def load_cert_file(path: Path):
    data = path.read_bytes()

    try:
        if b"-----BEGIN CERTIFICATE-----" in data:
            return [cx509.load_pem_x509_certificate(data)]
    except Exception:
        pass

    try:
        return [cx509.load_der_x509_certificate(data)]
    except Exception:
        pass

    return []


def is_ca(cert: cx509.Certificate) -> bool:
    try:
        bc = cert.extensions.get_extension_for_oid(
            ExtensionOID.BASIC_CONSTRAINTS
        ).value
        return bool(bc.ca)
    except cx509.ExtensionNotFound:
        return False

@lru_cache(maxsize=1)
def _load_trusted_tpm_certs(folder: str) -> list:
    certs = []
    base = Path(folder)

    paths = [base] if base.is_file() else base.rglob("*")

    for path in paths:
        if not path.is_file():
            continue

        for cert in load_cert_file(path):
            if is_ca(cert):
                certs.append((cert, str(path)))

    return certs


def is_directly_issued_by_cert_in_folder(cert: cx509.Certificate, folder: str):
    if cert is None:
        return False, None, None
    for candidate, path in _load_trusted_tpm_certs(folder):
        if cert.issuer != candidate.subject:
            continue
        try:
            cert.verify_directly_issued_by(candidate)
            return True, candidate, path
        except Exception:
            pass
    return False, None, None


def is_client_certificate_valid_for_ca_reference(
    client_cert,
    ca_reference,
    template_oid= None,
) -> bool:
    """Validate a TLS client certificate against one or more CA references.

    ``client_cert`` may be a cryptography Certificate, PEM bytes, or the
    URL-escaped PEM value received in ``X-Ssl-Client-Cert``.

    ``ca_reference`` may be a resolved CA dict, a CA id/display name/refid,
    or a list of those values. String/integer references are resolved from
    ``current_app.confadcs``. A matching CA must explicitly allow X.509
    authentication, directly issue the client certificate, and its configured
    CRL must not revoke the certificate. If the client certificate carries a
    BasicConstraints extension, it must have ``ca=False``. A missing
    BasicConstraints extension is accepted. When ``template_oid`` is provided,
    the certificate must also contain that template OID.
    """
    if isinstance(client_cert, cx509.Certificate):
        cert = client_cert
    else:
        try:
            if isinstance(client_cert, str):
                client_cert = unquote(client_cert).encode("utf-8")
            elif isinstance(client_cert, bytearray):
                client_cert = bytes(client_cert)

            if not isinstance(client_cert, bytes):
                return False

            cert = cx509.load_pem_x509_certificate(client_cert)
        except (TypeError, ValueError):
            return False

    now = datetime.now(timezone.utc)
    try:
        not_before = cert.not_valid_before_utc
        not_after = cert.not_valid_after_utc
    except AttributeError:
        # Compatibility with older cryptography releases.
        not_before = cert.not_valid_before.replace(tzinfo=timezone.utc)
        not_after = cert.not_valid_after.replace(tzinfo=timezone.utc)

    if now < not_before or now > not_after:
        return False

    try:
        basic_constraints = cert.extensions.get_extension_for_class(
            cx509.BasicConstraints
        ).value
    except cx509.ExtensionNotFound:
        basic_constraints = None

    if basic_constraints is not None and basic_constraints.ca:
        return False

    if template_oid and not _cert_has_template_oid(cert, template_oid):
        return False

    if isinstance(ca_reference, (list, tuple, set)):
        references = ca_reference
    else:
        references = [ca_reference]

    conf = None

    for reference in references:
        if isinstance(reference, dict):
            ca = reference
        else:
            if conf is None:
                try:
                    from flask import current_app
                    conf = current_app.confadcs
                except (ImportError, RuntimeError, AttributeError):
                    return False

            if isinstance(reference, int):
                ca = (conf.get("cas_by_refid") or {}).get(reference)
            elif isinstance(reference, str):
                ca = (
                    (conf.get("cas_by_id") or {}).get(reference)
                    or (conf.get("cas_by_display_name") or {}).get(reference)
                )
            else:
                ca = None

        if not isinstance(ca, dict):
            continue

        if not any(
            (entry.get("method") or "").strip().lower() == "x509"
            for entry in (ca.get("auth_methods") or [])
        ):
            continue

        ca_cert_path = ca.get("pem", {}).get("certificate_path_pem")
        if not ca_cert_path:
            continue

        issued, issuer_cert, _issuer_path = is_directly_issued_by_cert_in_folder(
            cert,
            ca_cert_path,
        )
        if not issued or issuer_cert is None:
            continue

        if is_certificate_revoked_by_crl(
            cert,
            ca.get("crl", {}).get("path_crl"),
            issuer_cert=issuer_cert,
        ):
            continue

        return True

    return False

# -----------------------------------------------------------------------------
# ML-DSA / raw SubjectPublicKeyInfo helpers for certificate issuance callbacks
# -----------------------------------------------------------------------------
from asn1crypto import algos as a_algos
from asn1crypto import keys as a_keys

ML_DSA_PUBLIC_KEY_OIDS = {
    "2.16.840.1.101.3.4.3.17",  # ML-DSA-44
    "2.16.840.1.101.3.4.3.18",  # ML-DSA-65
    "2.16.840.1.101.3.4.3.19",  # ML-DSA-87
}


def _asn1_time(dt):
    if dt.tzinfo is None:
        dt = dt.replace(tzinfo=timezone.utc)
    else:
        dt = dt.astimezone(timezone.utc)

    if dt.year < 2050:
        return a_x509.Time({"utc_time": a_core.UTCTime(dt)})
    return a_x509.Time({"general_time": a_core.GeneralizedTime(dt)})



def _der_read_len(data: bytes, offset: int):
    first = data[offset]
    offset += 1

    if first < 0x80:
        return first, offset

    n = first & 0x7F
    if n == 0:
        raise ValueError("Indefinite DER length is not allowed")

    length = int.from_bytes(data[offset:offset + n], "big")
    offset += n
    return length, offset


def _der_read_tlv(data: bytes, offset: int = 0):
    start = offset
    tag = data[offset]
    offset += 1

    # High-tag-number form, unlikely here but supported.
    if (tag & 0x1F) == 0x1F:
        while data[offset] & 0x80:
            offset += 1
        offset += 1

    length, content_start = _der_read_len(data, offset)
    end = content_start + length

    if end > len(data):
        raise ValueError("Invalid DER length")

    return tag, start, content_start, end


def _der_children(data: bytes, seq_start: int = 0):
    tag, _start, content_start, end = _der_read_tlv(data, seq_start)

    if tag != 0x30:
        raise ValueError(f"Expected SEQUENCE, got tag 0x{tag:02x}")

    children = []
    offset = content_start

    while offset < end:
        c_tag, c_start, c_content_start, c_end = _der_read_tlv(data, offset)
        children.append((c_tag, c_start, c_content_start, c_end))
        offset = c_end

    if offset != end:
        raise ValueError("Invalid DER sequence boundaries")

    return children


def _raw_x509_extension(oid: str, critical: bool, extn_value_der: bytes) -> a_x509.Extension:
    """
    Build Extension ::= SEQUENCE {
      extnID OBJECT IDENTIFIER,
      critical BOOLEAN DEFAULT FALSE,
      extnValue OCTET STRING -- containing DER of the real extension value
    }

    Construct this using raw DER to prevent asn1crypto from parsing
    certain recognized extensions.
    """
    contents = a_core.ObjectIdentifier(oid).dump()

    if critical:
        contents += a_core.Boolean(True).dump()

    contents += a_core.OctetString(extn_value_der).dump()

    return a_x509.Extension.load(_der_wrap_sequence(contents))


def _csr_spki_from_der(csr_der: bytes):
    """
    Returns:
      - spki_obj: raw ASN.1 SubjectPublicKeyInfo, usable in TbsCertificate
      - oid: public-key algorithm OID
      - ski: SubjectKeyIdentifier calculated from the raw BIT STRING

    Avoids typed public-key parsing, preserving ML-DSA compatibility.
    """

    # CertificationRequest ::= SEQUENCE {
    #   certificationRequestInfo CertificationRequestInfo,
    #   signatureAlgorithm AlgorithmIdentifier,
    #   signature BIT STRING
    # }
    csr_children = _der_children(csr_der, 0)
    if len(csr_children) < 1:
        raise ValueError("Invalid CSR: missing CertificationRequestInfo")

    _, cri_start, _, cri_end = csr_children[0]
    cri_der = csr_der[cri_start:cri_end]

    # CertificationRequestInfo ::= SEQUENCE {
    #   version INTEGER,
    #   subject Name,
    #   subjectPKInfo SubjectPublicKeyInfo,
    #   attributes [0] ...
    # }
    cri_children = _der_children(cri_der, 0)
    if len(cri_children) < 3:
        raise ValueError("Invalid CSR: missing subjectPKInfo")

    _, spki_start, _, spki_end = cri_children[2]
    spki_der = cri_der[spki_start:spki_end]

    # SubjectPublicKeyInfo ::= SEQUENCE {
    #   algorithm AlgorithmIdentifier,
    #   subjectPublicKey BIT STRING
    # }
    spki_children = _der_children(spki_der, 0)
    if len(spki_children) < 2:
        raise ValueError("Invalid SPKI")

    _, alg_start, _, alg_end = spki_children[0]
    bitstr_tag, bitstr_start, bitstr_content_start, bitstr_end = spki_children[1]

    alg_der = spki_der[alg_start:alg_end]
    alg_children = _der_children(alg_der, 0)
    if len(alg_children) < 1:
        raise ValueError("Invalid AlgorithmIdentifier")

    oid_tag, oid_start, _, oid_end = alg_children[0]
    if oid_tag != 0x06:
        raise ValueError(f"Expected OBJECT IDENTIFIER, got tag 0x{oid_tag:02x}")

    oid_der = alg_der[oid_start:oid_end]
    oid = a_core.ObjectIdentifier.load(oid_der).dotted

    if bitstr_tag != 0x03 or spki_der[bitstr_start] != 0x03:
        raise ValueError("Invalid SPKI: subjectPublicKey is not a BIT STRING")

    bitstr_contents = spki_der[bitstr_content_start:bitstr_end]
    if not bitstr_contents:
        raise ValueError("Invalid BIT STRING")

    unused_bits = bitstr_contents[0]
    if unused_bits != 0:
        raise ValueError("Unsupported subjectPublicKey BIT STRING with unused bits")

    public_key_bytes = bitstr_contents[1:]
    ski = hashlib.sha1(public_key_bytes).digest()

    # Important: load the raw object without accessing spki_obj["algorithm"]
    # or spki_obj["public_key"], which would make asn1crypto attempt to parse ML-DSA.
    spki_obj = a_keys.PublicKeyInfo.load(spki_der)

    return spki_obj, oid, ski




def _aia_extension_der(url: str) -> bytes:
    aia = a_x509.AuthorityInfoAccessSyntax([
        {
            "access_method": "ca_issuers",
            "access_location": {
                "uniform_resource_identifier": url,
            },
        }
    ])
    return aia.dump()


def _cdp_extension_der(url: str) -> bytes:
    gn = a_x509.GeneralName(
        name="uniform_resource_identifier",
        value=url,
    )

    cdp = a_x509.CRLDistributionPoints([
        a_x509.DistributionPoint({
            "distribution_point": a_x509.DistributionPointName(
                name="full_name",
                value=a_x509.GeneralNames([gn]),
            ),
        })
    ])

    return cdp.dump()


def _mldsa_ca_signature_oid(priv):
    """Return the ML-DSA signatureAlgorithm OID for a CA private key, or None."""
    if mldsa is None:
        return None

    if isinstance(priv, mldsa.MLDSA44PrivateKey):
        return "2.16.840.1.101.3.4.3.17"

    if isinstance(priv, mldsa.MLDSA65PrivateKey):
        return "2.16.840.1.101.3.4.3.18"

    if isinstance(priv, mldsa.MLDSA87PrivateKey):
        return "2.16.840.1.101.3.4.3.19"

    return None


def _signature_algo_for_ca_key(priv):
    if isinstance(priv, rsa.RSAPrivateKey):
        return a_algos.SignedDigestAlgorithm({"algorithm": "sha256_rsa"})

    if isinstance(priv, ec.EllipticCurvePrivateKey):
        return a_algos.SignedDigestAlgorithm({"algorithm": "sha256_ecdsa"})

    if isinstance(priv, ed25519.Ed25519PrivateKey):
        return a_algos.SignedDigestAlgorithm({"algorithm": "ed25519"})

    if isinstance(priv, ed448.Ed448PrivateKey):
        return a_algos.SignedDigestAlgorithm({"algorithm": "ed448"})

    mldsa_oid = _mldsa_ca_signature_oid(priv)
    if mldsa_oid:
        # ML-DSA algorithm identifiers are single-OID AlgorithmIdentifier values
        # with absent parameters. asn1crypto preserves the dotted OID even when
        # it does not have a friendly name for it.
        return a_algos.SignedDigestAlgorithm({"algorithm": mldsa_oid})

    raise ValueError(f"Unsupported CA private key type: {type(priv).__name__}")


def _sign_tbs_with_ca_key(priv, tbs_der: bytes) -> bytes:
    if isinstance(priv, rsa.RSAPrivateKey):
        return priv.sign(tbs_der, padding.PKCS1v15(), hashes.SHA256())

    if isinstance(priv, ec.EllipticCurvePrivateKey):
        return priv.sign(tbs_der, ec.ECDSA(hashes.SHA256()))

    if isinstance(priv, (ed25519.Ed25519PrivateKey, ed448.Ed448PrivateKey)):
        return priv.sign(tbs_der)

    if _mldsa_ca_signature_oid(priv):
        # ML-DSA signs the message directly; do not pass a hash algorithm or
        # RSA/ECDSA padding object.
        return priv.sign(tbs_der)

    raise ValueError(f"Unsupported CA private key type: {type(priv).__name__}")

def send_mail(
    *,
    smtp_conf: dict,
    to,
    subject: str,
    body: str,
    html: str = None,
):
    """
    Send an email using the Python standard library.

    Supported security modes:
      - starttls
      - ssl
      - none
    """

    if not smtp_conf:
        raise ValueError("SMTP configuration is missing")

    if not smtp_conf.get("enabled", False):
        return False

    host = smtp_conf.get("host")

    if not host:
        raise ValueError("SMTP host is missing")

    port = int(smtp_conf.get("port", 587))
    security = str(
        smtp_conf.get("security", "starttls")
    ).strip().lower()

    timeout = float(smtp_conf.get("timeout", 10))

    username = smtp_conf.get("username")
    password = smtp_conf.get("password")

    # password_file takes precedence over password
    password_file = smtp_conf.get("password_file")

    if password_file:
        try:
            with open(
                password_file,
                "r",
                encoding="utf-8",
            ) as f:
                password = f.readline().strip()

        except OSError as exc:
            raise ValueError(
                f"Unable to read SMTP password file "
                f"'{password_file}': {exc}"
            ) from exc

        if not password:
            raise ValueError(
                f"SMTP password file is empty: {password_file}"
            )

    sender = smtp_conf.get("sender") or username
    sender_name = smtp_conf.get("sender_name")

    if not sender:
        raise ValueError("SMTP sender is missing")

    # Accept either:
    # to="admin@domain.lan"
    # or:
    # to=["admin1@domain.lan", "admin2@domain.lan"]
    if isinstance(to, str):
        recipients = [to]
    else:
        recipients = list(to or [])

    recipients = [
        str(recipient).strip()
        for recipient in recipients
        if recipient
    ]

    if not recipients:
        raise ValueError("No email recipient specified")

    message = EmailMessage()

    if sender_name:
        message["From"] = formataddr(
            (sender_name, sender)
        )
    else:
        message["From"] = sender

    message["To"] = ", ".join(recipients)
    message["Subject"] = subject

    message.set_content(body)

    if html:
        message.add_alternative(
            html,
            subtype="html",
        )

    ca_file = smtp_conf.get("ca_file")

    ssl_context = ssl.create_default_context(
        cafile=ca_file if ca_file else None
    )

    #
    # SMTPS / implicit SSL
    #
    if security == "ssl":

        with smtplib.SMTP_SSL(
            host,
            port,
            timeout=timeout,
            context=ssl_context,
        ) as smtp:

            if username:
                if password is None:
                    raise ValueError(
                        "SMTP password is missing"
                    )

                smtp.login(
                    username,
                    password,
                )

            smtp.send_message(
                message,
                from_addr=sender,
                to_addrs=recipients,
            )

    #
    # SMTP / STARTTLS
    #
    else:

        with smtplib.SMTP(
            host,
            port,
            timeout=timeout,
        ) as smtp:

            smtp.ehlo()

            if security == "starttls":

                smtp.starttls(
                    context=ssl_context
                )

                smtp.ehlo()

            elif security != "none":

                raise ValueError(
                    f"Unknown SMTP security mode: "
                    f"{security}"
                )

            if username:
                if password is None:
                    raise ValueError(
                        "SMTP password is missing"
                    )

                smtp.login(
                    username,
                    password,
                )

            smtp.send_message(
                message,
                from_addr=sender,
                to_addrs=recipients,
            )

    return True

def get_default_realm():
    name = gssapi.Name(
        "dummy",
        name_type=gssapi.NameType.kerberos_principal,
    )
    principal = name.canonicalize(gssapi.MechType.kerberos)
    return str(principal).rsplit("@", 1)[1].upper()