diff --git a/libraries/microsoft-agents-authentication-msal/microsoft_agents/authentication/msal/_certificate_store.py b/libraries/microsoft-agents-authentication-msal/microsoft_agents/authentication/msal/_certificate_store.py new file mode 100644 index 000000000..04b7d421b --- /dev/null +++ b/libraries/microsoft-agents-authentication-msal/microsoft_agents/authentication/msal/_certificate_store.py @@ -0,0 +1,879 @@ +# Copyright (c) Microsoft Corporation. All rights reserved. +# Licensed under the MIT License. + +from __future__ import annotations + +import base64 +import ctypes +import hashlib +import json +import sys +import time +import uuid +import weakref + +_STORE_NAMES = { + "addressbook": "AddressBook", + "authroot": "AuthRoot", + "certificateauthority": "CA", + "disallowed": "Disallowed", + "my": "My", + "root": "Root", + "trustedpeople": "TrustedPeople", + "trustedpublisher": "TrustedPublisher", +} + +DWORD = ctypes.c_uint32 +LONG = ctypes.c_int32 +BOOL = ctypes.c_int32 +BYTE = ctypes.c_ubyte +HCERTSTORE = ctypes.c_void_p +HCRYPTKEY = ctypes.c_size_t +HCRYPTHASH = ctypes.c_size_t +HCRYPTPROV_OR_NCRYPT_KEY_HANDLE = ctypes.c_size_t + +X509_ASN_ENCODING = 0x00000001 +PKCS_7_ASN_ENCODING = 0x00010000 +CERT_ENCODING = X509_ASN_ENCODING | PKCS_7_ASN_ENCODING + +CERT_FIND_SUBJECT_STR_W = 0x00080007 +CERT_CHAIN_POLICY_BASE = 1 +CRYPT_ACQUIRE_PREFER_NCRYPT_KEY_FLAG = 0x00020000 + +AT_KEYEXCHANGE = 1 +AT_SIGNATURE = 2 +CERT_NCRYPT_KEY_SPEC = 0xFFFFFFFF + +KP_ALGID = 7 +CALG_RSA_SIGN = 0x00002400 +CALG_RSA_KEYX = 0x0000A400 + +CALG_SHA_256 = 0x0000800C +HP_HASHVAL = 0x0002 + +NCRYPT_ALGORITHM_GROUP_PROPERTY = "Algorithm Group" +NCRYPT_RSA_ALGORITHM_GROUP = "RSA" + +NCRYPT_PAD_PKCS1_FLAG = 0x00000002 +NCRYPT_PAD_PSS_FLAG = 0x00000008 +BCRYPT_SHA256_ALGORITHM = "SHA256" +SHA256_DIGEST_LENGTH = 32 + +_CLIENT_ASSERTION_LIFETIME_SECONDS = 60 * 10 + + +class _CERT_CONTEXT(ctypes.Structure): + pass + + +PCCERT_CONTEXT = ctypes.POINTER(_CERT_CONTEXT) + +_CERT_CONTEXT._fields_ = [ + ("dwCertEncodingType", DWORD), + ("pbCertEncoded", ctypes.POINTER(BYTE)), + ("cbCertEncoded", DWORD), + ("pCertInfo", ctypes.c_void_p), + ("hCertStore", HCERTSTORE), +] + + +class _CERT_ENHKEY_USAGE(ctypes.Structure): + _fields_ = [ + ("cUsageIdentifier", DWORD), + ("rgpszUsageIdentifier", ctypes.POINTER(ctypes.c_char_p)), + ] + + +class _CERT_USAGE_MATCH(ctypes.Structure): + _fields_ = [ + ("dwType", DWORD), + ("Usage", _CERT_ENHKEY_USAGE), + ] + + +class _CERT_CHAIN_PARA(ctypes.Structure): + _fields_ = [ + ("cbSize", DWORD), + ("RequestedUsage", _CERT_USAGE_MATCH), + ("RequestedIssuancePolicy", _CERT_USAGE_MATCH), + ("dwUrlRetrievalTimeout", DWORD), + ("fCheckRevocationFreshnessTime", BOOL), + ("dwRevocationFreshnessTime", DWORD), + ("pftCacheResync", ctypes.c_void_p), + ("pStrongSignPara", ctypes.c_void_p), + ("dwStrongSignFlags", DWORD), + ] + + +class _CERT_CHAIN_POLICY_PARA(ctypes.Structure): + _fields_ = [ + ("cbSize", DWORD), + ("dwFlags", DWORD), + ("pvExtraPolicyPara", ctypes.c_void_p), + ] + + +class _CERT_CHAIN_POLICY_STATUS(ctypes.Structure): + _fields_ = [ + ("cbSize", DWORD), + ("dwError", DWORD), + ("lChainIndex", LONG), + ("lElementIndex", LONG), + ("pvExtraPolicyStatus", ctypes.c_void_p), + ] + + +class _BCRYPT_PSS_PADDING_INFO(ctypes.Structure): + _fields_ = [ + ("pszAlgId", ctypes.c_wchar_p), + ("cbSalt", DWORD), + ] + + +class _BCRYPT_PKCS1_PADDING_INFO(ctypes.Structure): + _fields_ = [ + ("pszAlgId", ctypes.c_wchar_p), + ] + + +def _configure_crypt32(crypt32) -> None: + crypt32.CertOpenSystemStoreW.argtypes = [ + ctypes.c_void_p, + ctypes.c_wchar_p, + ] + crypt32.CertOpenSystemStoreW.restype = HCERTSTORE + + crypt32.CertFindCertificateInStore.argtypes = [ + HCERTSTORE, + DWORD, + DWORD, + DWORD, + ctypes.c_wchar_p, + PCCERT_CONTEXT, + ] + crypt32.CertFindCertificateInStore.restype = PCCERT_CONTEXT + + crypt32.CertDuplicateCertificateContext.argtypes = [PCCERT_CONTEXT] + crypt32.CertDuplicateCertificateContext.restype = PCCERT_CONTEXT + + crypt32.CertFreeCertificateContext.argtypes = [PCCERT_CONTEXT] + crypt32.CertFreeCertificateContext.restype = BOOL + + crypt32.CertCloseStore.argtypes = [HCERTSTORE, DWORD] + crypt32.CertCloseStore.restype = BOOL + + crypt32.CertGetCertificateChain.argtypes = [ + ctypes.c_void_p, + PCCERT_CONTEXT, + ctypes.c_void_p, + HCERTSTORE, + ctypes.POINTER(_CERT_CHAIN_PARA), + DWORD, + ctypes.c_void_p, + ctypes.POINTER(ctypes.c_void_p), + ] + crypt32.CertGetCertificateChain.restype = BOOL + + crypt32.CertVerifyCertificateChainPolicy.argtypes = [ + ctypes.c_void_p, + ctypes.c_void_p, + ctypes.POINTER(_CERT_CHAIN_POLICY_PARA), + ctypes.POINTER(_CERT_CHAIN_POLICY_STATUS), + ] + crypt32.CertVerifyCertificateChainPolicy.restype = BOOL + + crypt32.CertFreeCertificateChain.argtypes = [ctypes.c_void_p] + crypt32.CertFreeCertificateChain.restype = None + + crypt32.CryptAcquireCertificatePrivateKey.argtypes = [ + PCCERT_CONTEXT, + DWORD, + ctypes.c_void_p, + ctypes.POINTER(HCRYPTPROV_OR_NCRYPT_KEY_HANDLE), + ctypes.POINTER(DWORD), + ctypes.POINTER(BOOL), + ] + crypt32.CryptAcquireCertificatePrivateKey.restype = BOOL + + +def _configure_ncrypt(ncrypt) -> None: + ncrypt.NCryptFreeObject.argtypes = [ + HCRYPTPROV_OR_NCRYPT_KEY_HANDLE, + ] + ncrypt.NCryptFreeObject.restype = LONG + + ncrypt.NCryptGetProperty.argtypes = [ + HCRYPTPROV_OR_NCRYPT_KEY_HANDLE, + ctypes.c_wchar_p, + ctypes.POINTER(BYTE), + DWORD, + ctypes.POINTER(DWORD), + DWORD, + ] + ncrypt.NCryptGetProperty.restype = LONG + + ncrypt.NCryptSignHash.argtypes = [ + HCRYPTPROV_OR_NCRYPT_KEY_HANDLE, + ctypes.c_void_p, + ctypes.POINTER(BYTE), + DWORD, + ctypes.POINTER(BYTE), + DWORD, + ctypes.POINTER(DWORD), + DWORD, + ] + ncrypt.NCryptSignHash.restype = LONG + + +def _configure_advapi32(advapi32) -> None: + advapi32.CryptReleaseContext.argtypes = [ + HCRYPTPROV_OR_NCRYPT_KEY_HANDLE, + DWORD, + ] + advapi32.CryptReleaseContext.restype = BOOL + + advapi32.CryptGetUserKey.argtypes = [ + HCRYPTPROV_OR_NCRYPT_KEY_HANDLE, + DWORD, + ctypes.POINTER(HCRYPTKEY), + ] + advapi32.CryptGetUserKey.restype = BOOL + + advapi32.CryptGetKeyParam.argtypes = [ + HCRYPTKEY, + DWORD, + ctypes.POINTER(BYTE), + ctypes.POINTER(DWORD), + DWORD, + ] + advapi32.CryptGetKeyParam.restype = BOOL + + advapi32.CryptDestroyKey.argtypes = [HCRYPTKEY] + advapi32.CryptDestroyKey.restype = BOOL + + advapi32.CryptCreateHash.argtypes = [ + HCRYPTPROV_OR_NCRYPT_KEY_HANDLE, + DWORD, + HCRYPTKEY, + DWORD, + ctypes.POINTER(HCRYPTHASH), + ] + advapi32.CryptCreateHash.restype = BOOL + + advapi32.CryptSetHashParam.argtypes = [ + HCRYPTHASH, + DWORD, + ctypes.POINTER(BYTE), + DWORD, + ] + advapi32.CryptSetHashParam.restype = BOOL + + advapi32.CryptSignHashW.argtypes = [ + HCRYPTHASH, + DWORD, + ctypes.c_wchar_p, + DWORD, + ctypes.POINTER(BYTE), + ctypes.POINTER(DWORD), + ] + advapi32.CryptSignHashW.restype = BOOL + + advapi32.CryptDestroyHash.argtypes = [HCRYPTHASH] + advapi32.CryptDestroyHash.restype = BOOL + + +def _normalize_store_name(store_name: str | None) -> str: + if not store_name: + return "My" + + return _STORE_NAMES.get(store_name.casefold(), store_name) + + +def _load_windows_apis(): + if sys.platform != "win32": + raise OSError( + "CertificateSubjectName authentication requires " + "the Windows certificate store." + ) + + crypt32 = ctypes.WinDLL("crypt32", use_last_error=True) + ncrypt = ctypes.WinDLL("ncrypt", use_last_error=True) + advapi32 = ctypes.WinDLL("advapi32", use_last_error=True) + + _configure_crypt32(crypt32) + _configure_ncrypt(ncrypt) + _configure_advapi32(advapi32) + + return crypt32, ncrypt, advapi32 + + +def _is_certificate_valid(crypt32, certificate: PCCERT_CONTEXT) -> bool: + chain_parameters = _CERT_CHAIN_PARA() + chain_parameters.cbSize = ctypes.sizeof(chain_parameters) + + chain_context = ctypes.c_void_p() + + if not crypt32.CertGetCertificateChain( + None, + certificate, + None, + None, + ctypes.byref(chain_parameters), + 0, + None, + ctypes.byref(chain_context), + ): + raise OSError("Failed to build certificate chain.") + + try: + policy_parameters = _CERT_CHAIN_POLICY_PARA() + policy_parameters.cbSize = ctypes.sizeof(policy_parameters) + + policy_status = _CERT_CHAIN_POLICY_STATUS() + policy_status.cbSize = ctypes.sizeof(policy_status) + + if not crypt32.CertVerifyCertificateChainPolicy( + ctypes.c_void_p(CERT_CHAIN_POLICY_BASE), + chain_context, + ctypes.byref(policy_parameters), + ctypes.byref(policy_status), + ): + raise OSError("Failed to verify certificate chain policy.") + + return policy_status.dwError == 0 + finally: + crypt32.CertFreeCertificateChain(chain_context) + + +def _find_certificate_context( + crypt32, + *, + subject_name: str, + store_name: str, + valid_only: bool, +) -> PCCERT_CONTEXT: + """Find a matching certificate and return the certificate's duplicated context.""" + normalized_store_name = _normalize_store_name(store_name) + + store = crypt32.CertOpenSystemStoreW(None, normalized_store_name) + if not store: + raise OSError(f"Failed to open certificate store '{normalized_store_name}'.") + + certificate = None + try: + while True: + certificate = crypt32.CertFindCertificateInStore( + store, + CERT_ENCODING, + 0, + CERT_FIND_SUBJECT_STR_W, + subject_name, + certificate, + ) + + if not certificate: + raise LookupError( + f"No certificate matching subject '{subject_name}' " + f"was found in store '{normalized_store_name}'." + ) + + if valid_only and not _is_certificate_valid(crypt32, certificate): + continue + + duplicate = crypt32.CertDuplicateCertificateContext(certificate) + if not duplicate: + raise OSError("Failed to duplicate certificate context.") + + return duplicate + finally: + if certificate: + crypt32.CertFreeCertificateContext(certificate) + + crypt32.CertCloseStore(store, 0) + + +def _get_certificate_raw_data(certificate: PCCERT_CONTEXT) -> bytes: + return ctypes.string_at( + certificate.contents.pbCertEncoded, + certificate.contents.cbCertEncoded, + ) + + +def _base64url_encode(value: bytes) -> str: + return base64.urlsafe_b64encode(value).rstrip(b"=").decode("ascii") + + +def _compute_certificate_thumbprint( + certificate: PCCERT_CONTEXT, + *, + use_sha2: bool, +) -> str: + raw_data = _get_certificate_raw_data(certificate) + + if use_sha2: + digest = hashlib.sha256(raw_data).digest() + else: + digest = hashlib.sha1(raw_data, usedforsecurity=False).digest() + + return _base64url_encode(digest) + + +def _acquire_private_key( + crypt32, + certificate: PCCERT_CONTEXT, +) -> tuple[int, int, bool]: + key_handle = HCRYPTPROV_OR_NCRYPT_KEY_HANDLE() + key_spec = DWORD() + caller_free = BOOL() + + if not crypt32.CryptAcquireCertificatePrivateKey( + certificate, + CRYPT_ACQUIRE_PREFER_NCRYPT_KEY_FLAG, + None, + ctypes.byref(key_handle), + ctypes.byref(key_spec), + ctypes.byref(caller_free), + ): + raise OSError("Failed to acquire certificate private key.") + + return key_handle.value, key_spec.value, bool(caller_free.value) + + +def _release_private_key( + ncrypt, + advapi32, + *, + key_handle: int, + key_spec: int, + caller_free: bool, +) -> None: + if not caller_free: + return + + if key_spec == CERT_NCRYPT_KEY_SPEC: + if ncrypt.NCryptFreeObject(key_handle) != 0: + raise OSError("Failed to release CNG private key.") + return + + if not advapi32.CryptReleaseContext(key_handle, 0): + raise OSError("Failed to release certificate private key provider.") + + +def _is_cng_key_rsa(ncrypt, key_handle: int) -> bool: + result_size = DWORD() + + if ( + ncrypt.NCryptGetProperty( + key_handle, + NCRYPT_ALGORITHM_GROUP_PROPERTY, + None, + 0, + ctypes.byref(result_size), + 0, + ) + != 0 + ): + raise OSError("Failed to get CNG private key algorithm.") + + buffer = (BYTE * result_size.value)() + + if ( + ncrypt.NCryptGetProperty( + key_handle, + NCRYPT_ALGORITHM_GROUP_PROPERTY, + buffer, + result_size.value, + ctypes.byref(result_size), + 0, + ) + != 0 + ): + raise OSError("Failed to get CNG private key algorithm.") + + algorithm_group = ( + bytes(buffer[: result_size.value]).decode("utf-16-le").rstrip("\0") + ) + + return algorithm_group == NCRYPT_RSA_ALGORITHM_GROUP + + +def _is_legacy_key_rsa( + advapi32, + *, + key_handle: int, + key_spec: int, +) -> bool: + user_key = HCRYPTKEY() + + if not advapi32.CryptGetUserKey( + key_handle, + key_spec, + ctypes.byref(user_key), + ): + raise OSError("Failed to access certificate private key.") + + try: + algorithm = DWORD() + algorithm_size = DWORD(ctypes.sizeof(algorithm)) + + if not advapi32.CryptGetKeyParam( + user_key, + KP_ALGID, + ctypes.cast(ctypes.byref(algorithm), ctypes.POINTER(BYTE)), + ctypes.byref(algorithm_size), + 0, + ): + raise OSError("Failed to get certificate private key algorithm.") + except BaseException: + advapi32.CryptDestroyKey(user_key.value) + raise + + if not advapi32.CryptDestroyKey(user_key.value): + raise OSError("Failed to release certificate private key handle.") + + return algorithm.value in (CALG_RSA_SIGN, CALG_RSA_KEYX) + + +def _is_private_key_rsa( + ncrypt, + advapi32, + *, + key_handle: int, + key_spec: int, +) -> bool: + if key_spec == CERT_NCRYPT_KEY_SPEC: + return _is_cng_key_rsa(ncrypt, key_handle) + + return _is_legacy_key_rsa( + advapi32, + key_handle=key_handle, + key_spec=key_spec, + ) + + +def _sign_hash_with_cng( + ncrypt, + *, + key_handle: int, + digest: bytes, +) -> bytes: + if len(digest) != SHA256_DIGEST_LENGTH: + raise ValueError("PS256 signing requires a SHA-256 digest.") + + padding_info = _BCRYPT_PSS_PADDING_INFO( + pszAlgId=BCRYPT_SHA256_ALGORITHM, + cbSalt=SHA256_DIGEST_LENGTH, + ) + + digest_buffer = (BYTE * len(digest)).from_buffer_copy(digest) + signature_size = DWORD() + + if ( + ncrypt.NCryptSignHash( + key_handle, + ctypes.byref(padding_info), + digest_buffer, + len(digest), + None, + 0, + ctypes.byref(signature_size), + NCRYPT_PAD_PSS_FLAG, + ) + != 0 + ): + raise OSError("Failed to determine CNG signature size.") + + signature = (BYTE * signature_size.value)() + + if ( + ncrypt.NCryptSignHash( + key_handle, + ctypes.byref(padding_info), + digest_buffer, + len(digest), + signature, + signature_size.value, + ctypes.byref(signature_size), + NCRYPT_PAD_PSS_FLAG, + ) + != 0 + ): + raise OSError("Failed to sign with CNG private key.") + + return bytes(signature[: signature_size.value]) + + +def _sign_hash_with_cng_pkcs1( + ncrypt, + *, + key_handle: int, + digest: bytes, +) -> bytes: + if len(digest) != SHA256_DIGEST_LENGTH: + raise ValueError("RS256 signing requires a SHA-256 digest.") + + padding_info = _BCRYPT_PKCS1_PADDING_INFO( + pszAlgId=BCRYPT_SHA256_ALGORITHM, + ) + + digest_buffer = (BYTE * len(digest)).from_buffer_copy(digest) + signature_size = DWORD() + + if ( + ncrypt.NCryptSignHash( + key_handle, + ctypes.byref(padding_info), + digest_buffer, + len(digest), + None, + 0, + ctypes.byref(signature_size), + NCRYPT_PAD_PKCS1_FLAG, + ) + != 0 + ): + raise OSError("Failed to determine CNG PKCS#1 signature size.") + + signature = (BYTE * signature_size.value)() + + if ( + ncrypt.NCryptSignHash( + key_handle, + ctypes.byref(padding_info), + digest_buffer, + len(digest), + signature, + signature_size.value, + ctypes.byref(signature_size), + NCRYPT_PAD_PKCS1_FLAG, + ) + != 0 + ): + raise OSError("Failed to sign with CNG PKCS#1 private key.") + + return bytes(signature[: signature_size.value]) + + +def _sign_hash_with_legacy_csp( + advapi32, + *, + key_handle: int, + key_spec: int, + digest: bytes, +) -> bytes: + if len(digest) != SHA256_DIGEST_LENGTH: + raise ValueError("RS256 signing requires a SHA-256 digest.") + + hash_handle = HCRYPTHASH() + + if not advapi32.CryptCreateHash( + key_handle, + CALG_SHA_256, + 0, + 0, + ctypes.byref(hash_handle), + ): + raise OSError("Failed to create CSP SHA-256 hash.") + + try: + digest_buffer = (BYTE * len(digest)).from_buffer_copy(digest) + + if not advapi32.CryptSetHashParam( + hash_handle.value, + HP_HASHVAL, + digest_buffer, + 0, + ): + raise OSError("Failed to set CSP SHA-256 hash value.") + + signature_size = DWORD() + + if not advapi32.CryptSignHashW( + hash_handle.value, + key_spec, + None, + 0, + None, + ctypes.byref(signature_size), + ): + raise OSError("Failed to determine CSP RSA signature size.") + + signature = (BYTE * signature_size.value)() + + if not advapi32.CryptSignHashW( + hash_handle.value, + key_spec, + None, + 0, + signature, + ctypes.byref(signature_size), + ): + raise OSError("Failed to sign with CSP RSA private key.") + + result = bytes(signature[: signature_size.value])[::-1] + except BaseException: + advapi32.CryptDestroyHash(hash_handle.value) + raise + + if not advapi32.CryptDestroyHash(hash_handle.value): + raise OSError("Failed to release CSP hash handle.") + + return result + + +class _CertificateStoreClientAssertion: + """Creates MSAL client assertions backed by a certificate store certificate.""" + + def __init__( + self, + *, + subject_name: str, + store_name: str, + valid_only: bool, + send_x5c: bool, + client_id: str, + ): + self._crypt32, self._ncrypt, self._advapi32 = _load_windows_apis() + + self._certificate = _find_certificate_context( + self._crypt32, + subject_name=subject_name, + store_name=store_name, + valid_only=valid_only, + ) + self._certificate_finalizer = weakref.finalize( + self, + self._crypt32.CertFreeCertificateContext, + self._certificate, + ) + + self._raw_certificate = _get_certificate_raw_data(self._certificate) + self._sha1_thumbprint = _compute_certificate_thumbprint( + self._certificate, + use_sha2=False, + ) + self._sha256_thumbprint = _compute_certificate_thumbprint( + self._certificate, + use_sha2=True, + ) + + self._send_x5c = send_x5c + self._client_id = client_id + self._audience: str | None = None + + def bind_audience(self, audience: str) -> None: + """Bind the MSAL-resolved token endpoint used as the assertion audience.""" + if self._audience is not None and self._audience != audience: + raise RuntimeError("Certificate assertion audience is already bound.") + + self._audience = audience + + def __call__(self) -> str: + if self._audience is None: + raise RuntimeError( + "Certificate assertion audience has not been initialized." + ) + + valid_from = int(time.time()) + payload = { + "aud": self._audience, + "iss": self._client_id, + "sub": self._client_id, + "nbf": str(valid_from), + "exp": str(valid_from + _CLIENT_ASSERTION_LIFETIME_SECONDS), + "jti": str(uuid.uuid4()), + } + + key_handle, key_spec, caller_free = _acquire_private_key( + self._crypt32, + self._certificate, + ) + + try: + if not _is_private_key_rsa( + self._ncrypt, + self._advapi32, + key_handle=key_handle, + key_spec=key_spec, + ): + raise ValueError( + "The provided certificate is not of type RSA. " + "Please use a certificate of type RSA." + ) + + is_cng_key = key_spec == CERT_NCRYPT_KEY_SPEC + + if is_cng_key: + header = { + "alg": "PS256", + "typ": "JWT", + "x5t#S256": self._sha256_thumbprint, + } + else: + header = { + "alg": "RS256", + "typ": "JWT", + "x5t": self._sha1_thumbprint, + } + + if self._send_x5c: + header["x5c"] = base64.b64encode(self._raw_certificate).decode("ascii") + + encoded_payload = _base64url_encode( + json.dumps(payload, separators=(",", ":")).encode("utf-8") + ) + encoded_header = _base64url_encode( + json.dumps(header, separators=(",", ":")).encode("utf-8") + ) + + token = f"{encoded_header}.{encoded_payload}" + digest = hashlib.sha256(token.encode("ascii")).digest() + + if is_cng_key: + try: + signature = _sign_hash_with_cng( + self._ncrypt, + key_handle=key_handle, + digest=digest, + ) + except OSError: + header = { + "alg": "RS256", + "typ": "JWT", + "x5t": self._sha1_thumbprint, + } + + if self._send_x5c: + header["x5c"] = base64.b64encode(self._raw_certificate).decode( + "ascii" + ) + + encoded_header = _base64url_encode( + json.dumps(header, separators=(",", ":")).encode("utf-8") + ) + token = f"{encoded_header}.{encoded_payload}" + digest = hashlib.sha256(token.encode("ascii")).digest() + + signature = _sign_hash_with_cng_pkcs1( + self._ncrypt, + key_handle=key_handle, + digest=digest, + ) + else: + signature = _sign_hash_with_legacy_csp( + self._advapi32, + key_handle=key_handle, + key_spec=key_spec, + digest=digest, + ) + + return f"{token}.{_base64url_encode(signature)}" + + finally: + _release_private_key( + self._ncrypt, + self._advapi32, + key_handle=key_handle, + key_spec=key_spec, + caller_free=caller_free, + ) diff --git a/libraries/microsoft-agents-authentication-msal/microsoft_agents/authentication/msal/msal_auth.py b/libraries/microsoft-agents-authentication-msal/microsoft_agents/authentication/msal/msal_auth.py index 15f54afa8..f40d1504c 100644 --- a/libraries/microsoft-agents-authentication-msal/microsoft_agents/authentication/msal/msal_auth.py +++ b/libraries/microsoft-agents-authentication-msal/microsoft_agents/authentication/msal/msal_auth.py @@ -32,6 +32,9 @@ ) from microsoft_agents.hosting.core.authorization.telemetry import spans from microsoft_agents.authentication.msal.errors import authentication_errors +from microsoft_agents.authentication.msal._certificate_store import ( + _CertificateStoreClientAssertion, +) logger = logging.getLogger(__name__) @@ -299,6 +302,7 @@ def _create_client_application( else: authority = MsalAuth._resolve_authority(self._msal_configuration, tenant_id) client_credential = None + certificate_assertion = None if self._msal_configuration.AUTH_TYPE == AuthTypes.client_secret: client_credential = self._msal_configuration.CLIENT_SECRET @@ -308,6 +312,22 @@ def _create_client_application( } if self._msal_configuration.SEND_X5C: client_credential["public_certificate"] = True + elif ( + self._msal_configuration.AUTH_TYPE == AuthTypes.certificate_subject_name + ): + assert self._msal_configuration.CERT_SUBJECT_NAME is not None + assert self._msal_configuration.CLIENT_ID is not None + + certificate_assertion = _CertificateStoreClientAssertion( + subject_name=self._msal_configuration.CERT_SUBJECT_NAME, + store_name=self._msal_configuration.CERT_STORE_NAME, + valid_only=self._msal_configuration.VALID_CERTIFICATE_ONLY, + send_x5c=self._msal_configuration.SEND_X5C, + client_id=self._msal_configuration.CLIENT_ID, + ) + client_credential = { + "client_assertion": certificate_assertion, + } elif self._msal_configuration.AUTH_TYPE == AuthTypes.federated_credentials: mi_client = ManagedIdentityClient( UserAssignedManagedIdentity( @@ -353,13 +373,20 @@ def get_assertion() -> str: str(authentication_errors.AuthenticationTypeNotSupported) ) - return ConfidentialClientApplication( + client_application = ConfidentialClientApplication( client_id=self._msal_configuration.CLIENT_ID, authority=authority, client_credential=client_credential, azure_region=MsalAuth._resolve_azure_region(self._msal_configuration), ) + if certificate_assertion is not None: + certificate_assertion.bind_audience( + client_application.authority.token_endpoint + ) + + return client_application + def _client_rep( self, tenant_id: str | None = None, instance_id: str | None = None ) -> str: diff --git a/libraries/microsoft-agents-hosting-core/microsoft_agents/hosting/core/authorization/agent_auth_configuration.py b/libraries/microsoft-agents-hosting-core/microsoft_agents/hosting/core/authorization/agent_auth_configuration.py index ad188d50d..b3f4cc0b5 100644 --- a/libraries/microsoft-agents-hosting-core/microsoft_agents/hosting/core/authorization/agent_auth_configuration.py +++ b/libraries/microsoft-agents-hosting-core/microsoft_agents/hosting/core/authorization/agent_auth_configuration.py @@ -25,6 +25,9 @@ "TENANTID", "CLIENTSECRET", "CERTPFXFILE", + "CERTSUBJECTNAME", + "CERTSTORENAME", + "VALIDCERTIFICATEONLY", "SENDX5C", "CONNECTIONNAME", "FEDERATEDCLIENTID", @@ -64,6 +67,9 @@ class AgentAuthConfiguration: AUTH_TYPE: The type of authentication to use (microsoft_agents.hosting.core.authorization.auth_types.AuthTypes). CLIENT_SECRET: The client secret for the Azure AD application (if using client secret authentication). CERT_PFX_FILE: The path to the PFX certificate file (if using certificate authentication). + CERT_SUBJECT_NAME: The subject employed to locate a certificate for certificate subject name authentication. + CERT_STORE_NAME: The certificate store to search when resolving a certificate (default "My"). + VALID_CERTIFICATE_ONLY: Requires the certificate to have a valid chain (default True). SEND_X5C: Whether to include the public certificate in the x5c header for certificate authentication. CONNECTION_NAME: The name of the connection FEDERATED_CLIENT_ID: The client ID for federated credentials (if using federated credentials authentication). @@ -93,6 +99,9 @@ class AgentAuthConfiguration: CLIENT_ID: str | None CLIENT_SECRET: str | None CERT_PFX_FILE: str | None + CERT_SUBJECT_NAME: str | None + CERT_STORE_NAME: str = "My" + VALID_CERTIFICATE_ONLY: bool = True SEND_X5C: bool = False CONNECTION_NAME: str | None FEDERATED_CLIENT_ID: str | None @@ -128,6 +137,9 @@ def __init__( client_secret: str | None = None, *, cert_pfx_file: str | None = None, + cert_subject_name: str | None = None, + cert_store_name: str | None = None, + valid_certificate_only: bool | None = None, connection_name: str | None = None, federated_client_id: str | None = None, authority: str | None = None, @@ -154,6 +166,19 @@ def __init__( self.TENANT_ID = tenant_id or kwargs.get("TENANTID", None) self.CLIENT_SECRET = client_secret or kwargs.get("CLIENTSECRET", None) self.CERT_PFX_FILE = cert_pfx_file or kwargs.get("CERTPFXFILE", None) + self.CERT_SUBJECT_NAME = cert_subject_name or kwargs.get( + "CERTSUBJECTNAME", None + ) + self.CERT_STORE_NAME = cert_store_name or kwargs.get("CERTSTORENAME", "My") + self.VALID_CERTIFICATE_ONLY = coerce_bool( + ( + valid_certificate_only + if valid_certificate_only is not None + else kwargs.get("VALIDCERTIFICATEONLY", True) + ), + default=True, + name="VALIDCERTIFICATEONLY", + ) self.SEND_X5C = coerce_bool( send_x5c if send_x5c is not None else kwargs.get("SENDX5C", False), default=False, @@ -256,6 +281,13 @@ def _validate(self) -> None: raise ValueError( "FEDERATED_TOKEN_FILE is required for workload_identity authentication." ) + if ( + self.AUTH_TYPE == AuthTypes.certificate_subject_name + and not self.CERT_SUBJECT_NAME + ): + raise ValueError( + "CERT_SUBJECT_NAME is required for certificate_subject_name authentication." + ) @property def ISSUERS(self) -> list[str]: diff --git a/tests/authentication_msal/test_certificate_store.py b/tests/authentication_msal/test_certificate_store.py new file mode 100644 index 000000000..1b0a153f4 --- /dev/null +++ b/tests/authentication_msal/test_certificate_store.py @@ -0,0 +1,1165 @@ +import base64 +import ctypes +import hashlib +import json +import uuid + +import pytest +from microsoft_agents.authentication.msal._certificate_store import ( + _CERT_CONTEXT, + AT_KEYEXCHANGE, + AT_SIGNATURE, + CALG_RSA_KEYX, + CALG_RSA_SIGN, + CERT_ENCODING, + CERT_FIND_SUBJECT_STR_W, + CERT_NCRYPT_KEY_SPEC, + _acquire_private_key, + _CertificateStoreClientAssertion, + _compute_certificate_thumbprint, + _find_certificate_context, + _get_certificate_raw_data, + _is_certificate_valid, + _is_cng_key_rsa, + _is_legacy_key_rsa, + _is_private_key_rsa, + _load_windows_apis, + _normalize_store_name, + _release_private_key, + _sign_hash_with_cng, + _sign_hash_with_cng_pkcs1, + _sign_hash_with_legacy_csp, +) + + +def test_load_windows_apis_raises_on_non_windows(monkeypatch): + monkeypatch.setattr("sys.platform", "linux") + + with pytest.raises( + OSError, + match="CertificateSubjectName authentication requires the Windows certificate store", + ): + _load_windows_apis() + + +@pytest.mark.parametrize( + ("store_name", "expected"), + [ + (None, "My"), + ("", "My"), + ("My", "My"), + ("my", "My"), + ("Root", "Root"), + ("ROOT", "Root"), + ("CertificateAuthority", "CA"), + ("certificateauthority", "CA"), + ("TrustedPeople", "TrustedPeople"), + ("invalid-store", "invalid-store"), + ("AddressBook", "AddressBook"), + ("authroot", "AuthRoot"), + ("Disallowed", "Disallowed"), + ("trustedpublisher", "TrustedPublisher"), + ], +) +def test_normalize_store_name(store_name, expected): + assert _normalize_store_name(store_name) == expected + + +def test_find_certificate_context_duplicates_certificate(mocker): + crypt32 = mocker.Mock() + + store = ctypes.c_void_p(1) + certificate = ctypes.pointer(_CERT_CONTEXT()) + duplicate = ctypes.pointer(_CERT_CONTEXT()) + + crypt32.CertOpenSystemStoreW.return_value = store + crypt32.CertFindCertificateInStore.return_value = certificate + crypt32.CertDuplicateCertificateContext.return_value = duplicate + + result = _find_certificate_context( + crypt32, + subject_name="test-agent", + store_name="my", + valid_only=False, + ) + + assert result is duplicate + crypt32.CertOpenSystemStoreW.assert_called_once_with(None, "My") + crypt32.CertFindCertificateInStore.assert_called_once_with( + store, + CERT_ENCODING, + 0, + CERT_FIND_SUBJECT_STR_W, + "test-agent", + None, + ) + crypt32.CertDuplicateCertificateContext.assert_called_once_with(certificate) + crypt32.CertGetCertificateChain.assert_not_called() + crypt32.CertFreeCertificateContext.assert_called_once_with(certificate) + crypt32.CertCloseStore.assert_called_once_with(store, 0) + + +def test_find_certificate_context_closes_store_when_not_found(mocker): + crypt32 = mocker.Mock() + store = ctypes.c_void_p(1) + + crypt32.CertOpenSystemStoreW.return_value = store + crypt32.CertFindCertificateInStore.return_value = None + + with pytest.raises(LookupError, match="test-agent"): + _find_certificate_context( + crypt32, + subject_name="test-agent", + store_name="My", + valid_only=False, + ) + + crypt32.CertFreeCertificateContext.assert_not_called() + crypt32.CertCloseStore.assert_called_once_with(store, 0) + + +def test_find_certificate_context_raises_when_store_cannot_be_opened(mocker): + crypt32 = mocker.Mock() + crypt32.CertOpenSystemStoreW.return_value = None + + with pytest.raises(OSError, match="Failed to open certificate store 'My'"): + _find_certificate_context( + crypt32, + subject_name="test-agent", + store_name="My", + valid_only=False, + ) + + crypt32.CertFindCertificateInStore.assert_not_called() + crypt32.CertCloseStore.assert_not_called() + + +def test_find_certificate_context_frees_resources_when_duplication_fails(mocker): + crypt32 = mocker.Mock() + + store = ctypes.c_void_p(1) + certificate = ctypes.pointer(_CERT_CONTEXT()) + + crypt32.CertOpenSystemStoreW.return_value = store + crypt32.CertFindCertificateInStore.return_value = certificate + crypt32.CertDuplicateCertificateContext.return_value = None + + with pytest.raises(OSError, match="Failed to duplicate certificate context"): + _find_certificate_context( + crypt32, + subject_name="test-agent", + store_name="My", + valid_only=False, + ) + + crypt32.CertFreeCertificateContext.assert_called_once_with(certificate) + crypt32.CertCloseStore.assert_called_once_with(store, 0) + + +def test_is_certificate_valid_when_chain_policy_succeeds(mocker): + crypt32 = mocker.Mock() + certificate = ctypes.pointer(_CERT_CONTEXT()) + + crypt32.CertGetCertificateChain.return_value = True + crypt32.CertVerifyCertificateChainPolicy.return_value = True + + assert _is_certificate_valid(crypt32, certificate) + + crypt32.CertFreeCertificateChain.assert_called_once() + + +def test_is_certificate_valid_when_chain_policy_rejects_certificate(mocker): + crypt32 = mocker.Mock() + certificate = ctypes.pointer(_CERT_CONTEXT()) + + crypt32.CertGetCertificateChain.return_value = True + + def set_policy_error(_, __, ___, policy_status): + policy_status._obj.dwError = 1 + return True + + crypt32.CertVerifyCertificateChainPolicy.side_effect = set_policy_error + + assert not _is_certificate_valid(crypt32, certificate) + + crypt32.CertFreeCertificateChain.assert_called_once() + + +def test_is_certificate_valid_raises_when_chain_cannot_be_built(mocker): + crypt32 = mocker.Mock() + certificate = ctypes.pointer(_CERT_CONTEXT()) + + crypt32.CertGetCertificateChain.return_value = False + + with pytest.raises(OSError, match="Failed to build certificate chain"): + _is_certificate_valid(crypt32, certificate) + + crypt32.CertVerifyCertificateChainPolicy.assert_not_called() + crypt32.CertFreeCertificateChain.assert_not_called() + + +def test_is_certificate_valid_raises_when_chain_policy_cannot_be_verified(mocker): + crypt32 = mocker.Mock() + certificate = ctypes.pointer(_CERT_CONTEXT()) + + crypt32.CertGetCertificateChain.return_value = True + crypt32.CertVerifyCertificateChainPolicy.return_value = False + + with pytest.raises(OSError, match="Failed to verify certificate chain policy"): + _is_certificate_valid(crypt32, certificate) + + crypt32.CertFreeCertificateChain.assert_called_once() + + +def test_find_certificate_context_validates_certificate_when_required(mocker): + crypt32 = mocker.Mock() + + store = ctypes.c_void_p(1) + certificate = ctypes.pointer(_CERT_CONTEXT()) + duplicate = ctypes.pointer(_CERT_CONTEXT()) + + crypt32.CertOpenSystemStoreW.return_value = store + crypt32.CertFindCertificateInStore.return_value = certificate + crypt32.CertDuplicateCertificateContext.return_value = duplicate + crypt32.CertGetCertificateChain.return_value = True + crypt32.CertVerifyCertificateChainPolicy.return_value = True + + result = _find_certificate_context( + crypt32, + subject_name="test-agent", + store_name="My", + valid_only=True, + ) + + assert result is duplicate + crypt32.CertGetCertificateChain.assert_called_once() + + +def test_find_certificate_context_skips_invalid_certificate(mocker): + crypt32 = mocker.Mock() + + store = ctypes.c_void_p(1) + invalid_certificate = ctypes.pointer(_CERT_CONTEXT()) + valid_certificate = ctypes.pointer(_CERT_CONTEXT()) + duplicate = ctypes.pointer(_CERT_CONTEXT()) + + crypt32.CertOpenSystemStoreW.return_value = store + crypt32.CertFindCertificateInStore.side_effect = [ + invalid_certificate, + valid_certificate, + ] + crypt32.CertDuplicateCertificateContext.return_value = duplicate + + crypt32.CertGetCertificateChain.return_value = True + + policy_errors = iter([1, 0]) + + def set_policy_error(_, __, ___, policy_status): + policy_status._obj.dwError = next(policy_errors) + return True + + crypt32.CertVerifyCertificateChainPolicy.side_effect = set_policy_error + + result = _find_certificate_context( + crypt32, + subject_name="test-agent", + store_name="My", + valid_only=True, + ) + + assert result is duplicate + + assert crypt32.CertFindCertificateInStore.call_count == 2 + + first_call = crypt32.CertFindCertificateInStore.call_args_list[0] + second_call = crypt32.CertFindCertificateInStore.call_args_list[1] + + assert first_call.args[-1] is None + assert second_call.args[-1] is invalid_certificate + + crypt32.CertDuplicateCertificateContext.assert_called_once_with(valid_certificate) + crypt32.CertFreeCertificateContext.assert_called_once_with(valid_certificate) + crypt32.CertCloseStore.assert_called_once_with(store, 0) + + +def test_find_certificate_context_raises_when_all_certificates_are_invalid(mocker): + crypt32 = mocker.Mock() + + store = ctypes.c_void_p(1) + invalid_certificate = ctypes.pointer(_CERT_CONTEXT()) + + crypt32.CertOpenSystemStoreW.return_value = store + crypt32.CertFindCertificateInStore.side_effect = [ + invalid_certificate, + None, + ] + crypt32.CertGetCertificateChain.return_value = True + + def set_policy_error(_, __, ___, policy_status): + policy_status._obj.dwError = 1 + return True + + crypt32.CertVerifyCertificateChainPolicy.side_effect = set_policy_error + + with pytest.raises(LookupError, match="test-agent"): + _find_certificate_context( + crypt32, + subject_name="test-agent", + store_name="My", + valid_only=True, + ) + + assert crypt32.CertFindCertificateInStore.call_count == 2 + + first_call = crypt32.CertFindCertificateInStore.call_args_list[0] + second_call = crypt32.CertFindCertificateInStore.call_args_list[1] + + assert first_call.args[-1] is None + assert second_call.args[-1] is invalid_certificate + + crypt32.CertDuplicateCertificateContext.assert_not_called() + crypt32.CertFreeCertificateContext.assert_not_called() + crypt32.CertCloseStore.assert_called_once_with(store, 0) + + +def test_acquire_private_key_returns_cng_key(mocker): + crypt32 = mocker.Mock() + certificate = ctypes.pointer(_CERT_CONTEXT()) + + def acquire_key(_, __, ___, key_handle, key_spec, caller_free): + key_handle._obj.value = 123 + key_spec._obj.value = 0xFFFFFFFF + caller_free._obj.value = True + return True + + crypt32.CryptAcquireCertificatePrivateKey.side_effect = acquire_key + + result = _acquire_private_key(crypt32, certificate) + + assert result == (123, 0xFFFFFFFF, True) + + +def test_acquire_private_key_returns_legacy_key(mocker): + crypt32 = mocker.Mock() + certificate = ctypes.pointer(_CERT_CONTEXT()) + + def acquire_key(_, __, ___, key_handle, key_spec, caller_free): + key_handle._obj.value = 456 + key_spec._obj.value = 2 + caller_free._obj.value = False + return True + + crypt32.CryptAcquireCertificatePrivateKey.side_effect = acquire_key + + result = _acquire_private_key(crypt32, certificate) + + assert result == (456, 2, False) + + +def test_acquire_private_key_raises_when_acquisition_fails(mocker): + crypt32 = mocker.Mock() + certificate = ctypes.pointer(_CERT_CONTEXT()) + + crypt32.CryptAcquireCertificatePrivateKey.return_value = False + + with pytest.raises(OSError, match="Failed to acquire certificate private key"): + _acquire_private_key(crypt32, certificate) + + +def test_release_private_key_releases_cng_key(mocker): + ncrypt = mocker.Mock() + advapi32 = mocker.Mock() + + ncrypt.NCryptFreeObject.return_value = 0 + + _release_private_key( + ncrypt, + advapi32, + key_handle=123, + key_spec=CERT_NCRYPT_KEY_SPEC, + caller_free=True, + ) + + ncrypt.NCryptFreeObject.assert_called_once_with(123) + advapi32.CryptReleaseContext.assert_not_called() + + +def test_release_private_key_releases_legacy_provider(mocker): + ncrypt = mocker.Mock() + advapi32 = mocker.Mock() + + advapi32.CryptReleaseContext.return_value = True + + _release_private_key( + ncrypt, + advapi32, + key_handle=456, + key_spec=AT_SIGNATURE, + caller_free=True, + ) + + advapi32.CryptReleaseContext.assert_called_once_with(456, 0) + ncrypt.NCryptFreeObject.assert_not_called() + + +def test_release_private_key_does_not_release_unowned_handle(mocker): + ncrypt = mocker.Mock() + advapi32 = mocker.Mock() + + _release_private_key( + ncrypt, + advapi32, + key_handle=123, + key_spec=CERT_NCRYPT_KEY_SPEC, + caller_free=False, + ) + + ncrypt.NCryptFreeObject.assert_not_called() + advapi32.CryptReleaseContext.assert_not_called() + + +def test_release_private_key_raises_when_cng_release_fails(mocker): + ncrypt = mocker.Mock() + advapi32 = mocker.Mock() + + ncrypt.NCryptFreeObject.return_value = 1 + + with pytest.raises(OSError, match="Failed to release CNG private key"): + _release_private_key( + ncrypt, + advapi32, + key_handle=123, + key_spec=CERT_NCRYPT_KEY_SPEC, + caller_free=True, + ) + + advapi32.CryptReleaseContext.assert_not_called() + + +def test_release_private_key_raises_when_legacy_release_fails(mocker): + ncrypt = mocker.Mock() + advapi32 = mocker.Mock() + + advapi32.CryptReleaseContext.return_value = False + + with pytest.raises( + OSError, + match="Failed to release certificate private key provider", + ): + _release_private_key( + ncrypt, + advapi32, + key_handle=456, + key_spec=AT_SIGNATURE, + caller_free=True, + ) + + ncrypt.NCryptFreeObject.assert_not_called() + + +def test_is_cng_key_rsa_returns_true_for_rsa(mocker): + ncrypt = mocker.Mock() + encoded_algorithm = "RSA\0".encode("utf-16-le") + + def get_property(_, __, buffer, ___, result_size, ____): + result_size._obj.value = len(encoded_algorithm) + + if buffer is not None: + for index, value in enumerate(encoded_algorithm): + buffer[index] = value + + return 0 + + ncrypt.NCryptGetProperty.side_effect = get_property + + assert _is_cng_key_rsa(ncrypt, 123) + + +def test_is_cng_key_rsa_returns_false_for_non_rsa(mocker): + ncrypt = mocker.Mock() + encoded_algorithm = "ECDSA\0".encode("utf-16-le") + + def get_property(_, __, buffer, ___, result_size, ____): + result_size._obj.value = len(encoded_algorithm) + + if buffer is not None: + for index, value in enumerate(encoded_algorithm): + buffer[index] = value + + return 0 + + ncrypt.NCryptGetProperty.side_effect = get_property + + assert not _is_cng_key_rsa(ncrypt, 123) + + +@pytest.mark.parametrize("algorithm", [CALG_RSA_SIGN, CALG_RSA_KEYX]) +def test_is_legacy_key_rsa_returns_true_for_rsa(mocker, algorithm): + advapi32 = mocker.Mock() + + def get_user_key(_, __, user_key): + user_key._obj.value = 789 + return True + + def get_key_param(_, __, algorithm_buffer, ___, ____): + ctypes.cast( + algorithm_buffer, + ctypes.POINTER(ctypes.c_uint32), + ).contents.value = algorithm + return True + + advapi32.CryptGetUserKey.side_effect = get_user_key + advapi32.CryptGetKeyParam.side_effect = get_key_param + advapi32.CryptDestroyKey.return_value = True + + assert _is_legacy_key_rsa( + advapi32, + key_handle=456, + key_spec=AT_SIGNATURE, + ) + + advapi32.CryptDestroyKey.assert_called_once_with(789) + + +def test_is_legacy_key_rsa_returns_false_for_non_rsa(mocker): + advapi32 = mocker.Mock() + + def get_user_key(_, __, user_key): + user_key._obj.value = 789 + return True + + def get_key_param(_, __, algorithm_buffer, ___, ____): + ctypes.cast( + algorithm_buffer, + ctypes.POINTER(ctypes.c_uint32), + ).contents.value = 0 + return True + + advapi32.CryptGetUserKey.side_effect = get_user_key + advapi32.CryptGetKeyParam.side_effect = get_key_param + advapi32.CryptDestroyKey.return_value = True + + assert not _is_legacy_key_rsa( + advapi32, + key_handle=456, + key_spec=AT_KEYEXCHANGE, + ) + + +def test_is_private_key_rsa_uses_cng_for_ncrypt_key(mocker): + ncrypt = mocker.Mock() + advapi32 = mocker.Mock() + + encoded_algorithm = "RSA\0".encode("utf-16-le") + + def get_property(_, __, buffer, ___, result_size, ____): + result_size._obj.value = len(encoded_algorithm) + if buffer is not None: + for index, value in enumerate(encoded_algorithm): + buffer[index] = value + return 0 + + ncrypt.NCryptGetProperty.side_effect = get_property + + assert _is_private_key_rsa( + ncrypt, + advapi32, + key_handle=123, + key_spec=CERT_NCRYPT_KEY_SPEC, + ) + + advapi32.CryptGetUserKey.assert_not_called() + + +def test_sign_hash_with_cng_returns_signature(mocker): + ncrypt = mocker.Mock() + digest = bytes(range(32)) + expected_signature = b"\x01\x02\x03\x04" + + def sign_hash( + _key_handle, + _padding_info, + _digest, + _digest_size, + signature, + _signature_buffer_size, + result_size, + _flags, + ): + result_size._obj.value = len(expected_signature) + + if signature is not None: + for index, value in enumerate(expected_signature): + signature[index] = value + + return 0 + + ncrypt.NCryptSignHash.side_effect = sign_hash + + result = _sign_hash_with_cng( + ncrypt, + key_handle=123, + digest=digest, + ) + + assert result == expected_signature + assert ncrypt.NCryptSignHash.call_count == 2 + + +def test_sign_hash_with_cng_requires_sha256_digest(mocker): + ncrypt = mocker.Mock() + + with pytest.raises(ValueError, match="SHA-256 digest"): + _sign_hash_with_cng( + ncrypt, + key_handle=123, + digest=b"invalid", + ) + + ncrypt.NCryptSignHash.assert_not_called() + + +def test_sign_hash_with_cng_raises_when_signing_fails(mocker): + ncrypt = mocker.Mock() + digest = bytes(32) + + def sign_hash( + _key_handle, + _padding_info, + _digest, + _digest_size, + signature, + _signature_buffer_size, + result_size, + _flags, + ): + if signature is None: + result_size._obj.value = 256 + return 0 + + return 1 + + ncrypt.NCryptSignHash.side_effect = sign_hash + + with pytest.raises(OSError, match="Failed to sign with CNG private key"): + _sign_hash_with_cng( + ncrypt, + key_handle=123, + digest=digest, + ) + + +def test_sign_hash_with_legacy_csp_returns_signature(mocker): + advapi32 = mocker.Mock() + digest = bytes(range(32)) + + native_signature = b"\x01\x02\x03\x04" + + def create_hash( + _provider_handle, + _algorithm, + _key, + _flags, + hash_handle, + ): + hash_handle._obj.value = 789 + return True + + def sign_hash( + _hash_handle, + _key_spec, + _description, + _flags, + signature, + signature_size, + ): + signature_size._obj.value = len(native_signature) + + if signature is not None: + for index, value in enumerate(native_signature): + signature[index] = value + + return True + + advapi32.CryptCreateHash.side_effect = create_hash + advapi32.CryptSetHashParam.return_value = True + advapi32.CryptSignHashW.side_effect = sign_hash + advapi32.CryptDestroyHash.return_value = True + + result = _sign_hash_with_legacy_csp( + advapi32, + key_handle=456, + key_spec=AT_SIGNATURE, + digest=digest, + ) + + assert result == native_signature[::-1] + assert advapi32.CryptSignHashW.call_count == 2 + advapi32.CryptDestroyHash.assert_called_once_with(789) + + +def test_sign_hash_with_legacy_csp_requires_sha256_digest(mocker): + advapi32 = mocker.Mock() + + with pytest.raises(ValueError, match="SHA-256 digest"): + _sign_hash_with_legacy_csp( + advapi32, + key_handle=456, + key_spec=AT_SIGNATURE, + digest=b"invalid", + ) + + advapi32.CryptCreateHash.assert_not_called() + + +def test_sign_hash_with_legacy_csp_raises_when_signing_fails(mocker): + advapi32 = mocker.Mock() + digest = bytes(32) + + def create_hash( + _provider_handle, + _algorithm, + _key, + _flags, + hash_handle, + ): + hash_handle._obj.value = 789 + return True + + def sign_hash( + _hash_handle, + _key_spec, + _description, + _flags, + signature, + signature_size, + ): + if signature is None: + signature_size._obj.value = 256 + return True + + return False + + advapi32.CryptCreateHash.side_effect = create_hash + advapi32.CryptSetHashParam.return_value = True + advapi32.CryptSignHashW.side_effect = sign_hash + advapi32.CryptDestroyHash.return_value = True + + with pytest.raises( + OSError, + match="Failed to sign with CSP RSA private key", + ): + _sign_hash_with_legacy_csp( + advapi32, + key_handle=456, + key_spec=AT_SIGNATURE, + digest=digest, + ) + + advapi32.CryptDestroyHash.assert_called_once_with(789) + + +def _create_certificate_context(raw_data: bytes): + encoded = (ctypes.c_ubyte * len(raw_data))(*raw_data) + + certificate = _CERT_CONTEXT() + certificate.pbCertEncoded = ctypes.cast( + encoded, + ctypes.POINTER(ctypes.c_ubyte), + ) + certificate.cbCertEncoded = len(encoded) + + return ctypes.pointer(certificate), encoded + + +def _create_certificate_store_assertion(mocker, *, send_x5c=False): + crypt32 = mocker.Mock() + ncrypt = mocker.Mock() + advapi32 = mocker.Mock() + + certificate, encoded = _create_certificate_context(b"\x01\x02\x03\x04") + + mocker.patch( + "microsoft_agents.authentication.msal._certificate_store." "_load_windows_apis", + return_value=(crypt32, ncrypt, advapi32), + ) + mocker.patch( + "microsoft_agents.authentication.msal._certificate_store." + "_find_certificate_context", + return_value=certificate, + ) + mocker.patch( + "microsoft_agents.authentication.msal._certificate_store." "weakref.finalize" + ) + + assertion = _CertificateStoreClientAssertion( + subject_name="test-agent", + store_name="My", + valid_only=True, + send_x5c=send_x5c, + client_id="test-client-id", + ) + + return assertion, crypt32, ncrypt, advapi32, encoded + + +def _decode_jwt_part(value: str): + padding = "=" * (-len(value) % 4) + return json.loads(base64.urlsafe_b64decode(value + padding)) + + +def test_get_certificate_raw_data(): + certificate, _encoded = _create_certificate_context(b"\x01\x02\x03\x04") + + assert _get_certificate_raw_data(certificate) == b"\x01\x02\x03\x04" + + +@pytest.mark.parametrize( + ("use_sha2", "expected"), + [ + (False, "EtraH_9NR4et4zMxRyAsO0Q-N28"), + (True, "n2SnR-G5fxMfq7a0Rylsm28CAeefs8U1bmx36JtqgGo"), + ], +) +def test_compute_certificate_thumbprint(use_sha2, expected): + certificate, _encoded = _create_certificate_context(b"\x01\x02\x03\x04") + + assert _compute_certificate_thumbprint(certificate, use_sha2=use_sha2) == expected + + +def test_sign_hash_with_cng_pkcs1_returns_signature(mocker): + ncrypt = mocker.Mock() + digest = bytes(range(32)) + expected_signature = b"\x01\x02\x03\x04" + + def sign_hash( + _key_handle, + _padding_info, + _digest, + _digest_size, + signature, + _signature_buffer_size, + result_size, + _flags, + ): + result_size._obj.value = len(expected_signature) + + if signature is not None: + for index, value in enumerate(expected_signature): + signature[index] = value + + return 0 + + ncrypt.NCryptSignHash.side_effect = sign_hash + + result = _sign_hash_with_cng_pkcs1( + ncrypt, + key_handle=123, + digest=digest, + ) + + assert result == expected_signature + assert ncrypt.NCryptSignHash.call_count == 2 + + +def test_sign_hash_with_cng_pkcs1_requires_sha256_digest(mocker): + ncrypt = mocker.Mock() + + with pytest.raises(ValueError, match="SHA-256 digest"): + _sign_hash_with_cng_pkcs1( + ncrypt, + key_handle=123, + digest=b"invalid", + ) + + ncrypt.NCryptSignHash.assert_not_called() + + +def test_sign_hash_with_cng_pkcs1_raises_when_signing_fails(mocker): + ncrypt = mocker.Mock() + digest = bytes(32) + + def sign_hash( + _key_handle, + _padding_info, + _digest, + _digest_size, + signature, + _signature_buffer_size, + result_size, + _flags, + ): + if signature is None: + result_size._obj.value = 256 + return 0 + + return 1 + + ncrypt.NCryptSignHash.side_effect = sign_hash + + with pytest.raises( + OSError, + match="Failed to sign with CNG PKCS#1 private key", + ): + _sign_hash_with_cng_pkcs1( + ncrypt, + key_handle=123, + digest=digest, + ) + + +def test_certificate_store_client_assertion_signs_with_cng(mocker): + assertion, _crypt32, ncrypt, advapi32, _encoded = ( + _create_certificate_store_assertion(mocker, send_x5c=True) + ) + + mocker.patch( + "microsoft_agents.authentication.msal._certificate_store." + "_acquire_private_key", + return_value=(123, CERT_NCRYPT_KEY_SPEC, True), + ) + mocker.patch( + "microsoft_agents.authentication.msal._certificate_store." + "_is_private_key_rsa", + return_value=True, + ) + sign_cng = mocker.patch( + "microsoft_agents.authentication.msal._certificate_store." + "_sign_hash_with_cng", + return_value=b"\x01\x02\x03\x04", + ) + sign_legacy = mocker.patch( + "microsoft_agents.authentication.msal._certificate_store." + "_sign_hash_with_legacy_csp" + ) + release_key = mocker.patch( + "microsoft_agents.authentication.msal._certificate_store." + "_release_private_key" + ) + mocker.patch( + "microsoft_agents.authentication.msal._certificate_store.time.time", + return_value=1000, + ) + mocker.patch( + "microsoft_agents.authentication.msal._certificate_store.uuid.uuid4", + return_value=uuid.UUID("12345678-1234-5678-1234-567812345678"), + ) + + assertion.bind_audience("https://login.example/token") + + result = assertion() + encoded_header, encoded_payload, encoded_signature = result.split(".") + + assert _decode_jwt_part(encoded_header) == { + "alg": "PS256", + "typ": "JWT", + "x5t#S256": "n2SnR-G5fxMfq7a0Rylsm28CAeefs8U1bmx36JtqgGo", + "x5c": "AQIDBA==", + } + assert _decode_jwt_part(encoded_payload) == { + "aud": "https://login.example/token", + "iss": "test-client-id", + "sub": "test-client-id", + "nbf": "1000", + "exp": "1600", + "jti": "12345678-1234-5678-1234-567812345678", + } + + expected_digest = hashlib.sha256( + f"{encoded_header}.{encoded_payload}".encode("ascii") + ).digest() + + sign_cng.assert_called_once_with( + ncrypt, + key_handle=123, + digest=expected_digest, + ) + sign_legacy.assert_not_called() + + assert ( + base64.urlsafe_b64decode( + encoded_signature + "=" * (-len(encoded_signature) % 4) + ) + == b"\x01\x02\x03\x04" + ) + + release_key.assert_called_once_with( + ncrypt, + advapi32, + key_handle=123, + key_spec=CERT_NCRYPT_KEY_SPEC, + caller_free=True, + ) + + +def test_certificate_store_client_assertion_signs_with_legacy_csp(mocker): + assertion, _crypt32, ncrypt, advapi32, _encoded = ( + _create_certificate_store_assertion(mocker) + ) + + mocker.patch( + "microsoft_agents.authentication.msal._certificate_store." + "_acquire_private_key", + return_value=(456, AT_SIGNATURE, False), + ) + mocker.patch( + "microsoft_agents.authentication.msal._certificate_store." + "_is_private_key_rsa", + return_value=True, + ) + sign_cng = mocker.patch( + "microsoft_agents.authentication.msal._certificate_store." "_sign_hash_with_cng" + ) + sign_legacy = mocker.patch( + "microsoft_agents.authentication.msal._certificate_store." + "_sign_hash_with_legacy_csp", + return_value=b"\x05\x06\x07\x08", + ) + release_key = mocker.patch( + "microsoft_agents.authentication.msal._certificate_store." + "_release_private_key" + ) + + assertion.bind_audience("https://login.example/token") + + result = assertion() + encoded_header, encoded_payload, _encoded_signature = result.split(".") + + assert _decode_jwt_part(encoded_header) == { + "alg": "RS256", + "typ": "JWT", + "x5t": "EtraH_9NR4et4zMxRyAsO0Q-N28", + } + + expected_digest = hashlib.sha256( + f"{encoded_header}.{encoded_payload}".encode("ascii") + ).digest() + + sign_legacy.assert_called_once_with( + advapi32, + key_handle=456, + key_spec=AT_SIGNATURE, + digest=expected_digest, + ) + sign_cng.assert_not_called() + + release_key.assert_called_once_with( + ncrypt, + advapi32, + key_handle=456, + key_spec=AT_SIGNATURE, + caller_free=False, + ) + + +def test_certificate_store_client_assertion_requires_audience(mocker): + assertion, _crypt32, _ncrypt, _advapi32, _encoded = ( + _create_certificate_store_assertion(mocker) + ) + + with pytest.raises( + RuntimeError, + match="audience has not been initialized", + ): + assertion() + + +def test_certificate_store_client_assertion_rejects_non_rsa_key(mocker): + assertion, _crypt32, ncrypt, advapi32, _encoded = ( + _create_certificate_store_assertion(mocker) + ) + + mocker.patch( + "microsoft_agents.authentication.msal._certificate_store." + "_acquire_private_key", + return_value=(123, CERT_NCRYPT_KEY_SPEC, True), + ) + mocker.patch( + "microsoft_agents.authentication.msal._certificate_store." + "_is_private_key_rsa", + return_value=False, + ) + release_key = mocker.patch( + "microsoft_agents.authentication.msal._certificate_store." + "_release_private_key" + ) + + assertion.bind_audience("https://login.example/token") + + with pytest.raises(ValueError, match="certificate is not of type RSA"): + assertion() + + release_key.assert_called_once_with( + ncrypt, + advapi32, + key_handle=123, + key_spec=CERT_NCRYPT_KEY_SPEC, + caller_free=True, + ) + + +def test_certificate_store_client_assertion_falls_back_to_cng_pkcs1(mocker): + assertion, _crypt32, ncrypt, advapi32, _encoded = ( + _create_certificate_store_assertion(mocker, send_x5c=True) + ) + + mocker.patch( + "microsoft_agents.authentication.msal._certificate_store." + "_acquire_private_key", + return_value=(123, CERT_NCRYPT_KEY_SPEC, True), + ) + mocker.patch( + "microsoft_agents.authentication.msal._certificate_store." + "_is_private_key_rsa", + return_value=True, + ) + sign_cng = mocker.patch( + "microsoft_agents.authentication.msal._certificate_store." + "_sign_hash_with_cng", + side_effect=OSError("Failed to sign with CNG private key."), + ) + sign_pkcs1 = mocker.patch( + "microsoft_agents.authentication.msal._certificate_store." + "_sign_hash_with_cng_pkcs1", + return_value=b"\x01\x02\x03\x04", + ) + sign_legacy = mocker.patch( + "microsoft_agents.authentication.msal._certificate_store." + "_sign_hash_with_legacy_csp" + ) + release_key = mocker.patch( + "microsoft_agents.authentication.msal._certificate_store." + "_release_private_key" + ) + + assertion.bind_audience("https://login.example/token") + + result = assertion() + encoded_header, encoded_payload, _encoded_signature = result.split(".") + + assert _decode_jwt_part(encoded_header) == { + "alg": "RS256", + "typ": "JWT", + "x5t": "EtraH_9NR4et4zMxRyAsO0Q-N28", + "x5c": "AQIDBA==", + } + + expected_digest = hashlib.sha256( + f"{encoded_header}.{encoded_payload}".encode("ascii") + ).digest() + + sign_cng.assert_called_once() + sign_pkcs1.assert_called_once_with( + ncrypt, + key_handle=123, + digest=expected_digest, + ) + sign_legacy.assert_not_called() + + release_key.assert_called_once_with( + ncrypt, + advapi32, + key_handle=123, + key_spec=CERT_NCRYPT_KEY_SPEC, + caller_free=True, + ) diff --git a/tests/authentication_msal/test_msal_auth.py b/tests/authentication_msal/test_msal_auth.py index d68e6b25b..1b9070f6d 100644 --- a/tests/authentication_msal/test_msal_auth.py +++ b/tests/authentication_msal/test_msal_auth.py @@ -321,6 +321,49 @@ def test_create_client_application_send_x5c_defaults_false(self, mocker): } +class TestMsalAuthCertificateSubjectName: + def test_create_client_application_certificate_subject_name(self, mocker): + config = AgentAuthConfiguration( + auth_type=AuthTypes.certificate_subject_name, + tenant_id="12345678-1234-1234-1234-123456789abc", + client_id="test-client-id", + cert_subject_name="CN=test-agent", + cert_store_name="Root", + valid_certificate_only=False, + send_x5c=True, + ) + + mock_credential_class = mocker.patch( + "microsoft_agents.authentication.msal.msal_auth." + "_CertificateStoreClientAssertion", + ) + mock_credential = mock_credential_class.return_value + + mock_cca = mocker.patch( + "microsoft_agents.authentication.msal.msal_auth." + "ConfidentialClientApplication" + ) + mock_cca.return_value.authority.token_endpoint = ( + "https://resolved.example/token" + ) + + MsalAuth(config)._create_client_application() + + mock_credential_class.assert_called_once_with( + subject_name="CN=test-agent", + store_name="Root", + valid_only=False, + send_x5c=True, + client_id="test-client-id", + ) + assert mock_cca.call_args.kwargs["client_credential"] == { + "client_assertion": mock_credential, + } + mock_credential.bind_audience.assert_called_once_with( + "https://resolved.example/token" + ) + + class TestMsalAuthWorkloadIdentity: def test_create_client_application_reads_projected_token(self, mocker, tmp_path): token_file = tmp_path / "workload-token" diff --git a/tests/hosting_core/test_auth_configuration.py b/tests/hosting_core/test_auth_configuration.py index b349a2c28..704742d0f 100644 --- a/tests/hosting_core/test_auth_configuration.py +++ b/tests/hosting_core/test_auth_configuration.py @@ -99,6 +99,9 @@ def test_empty_settings(self): assert auth_config.CLIENT_ID is None assert auth_config.CLIENT_SECRET is None assert auth_config.CERT_PFX_FILE is None + assert auth_config.CERT_SUBJECT_NAME is None + assert auth_config.CERT_STORE_NAME == "My" + assert auth_config.VALID_CERTIFICATE_ONLY is True assert auth_config.SEND_X5C is False assert auth_config.FEDERATED_CLIENT_ID is None assert auth_config.CONNECTION_NAME is None @@ -123,6 +126,13 @@ def test_empty_settings(self): "FEDERATED_TOKEN_FILE is required for " "workload_identity authentication.", ), + ( + AuthTypes.certificate_subject_name, + ( + "CERT_SUBJECT_NAME is required for " + "certificate_subject_name authentication." + ), + ), ], ) def test_auth_type_requires_credential_setting(self, auth_type, expected_message): @@ -141,6 +151,10 @@ def test_auth_type_requires_credential_setting(self, auth_type, expected_message AuthTypes.workload_identity, {"federated_token_file": "test-token-file"}, ), + ( + AuthTypes.certificate_subject_name, + {"cert_subject_name": "CN=test-agent"}, + ), ], ) def test_auth_type_accepts_required_credential_setting(self, auth_type, credential): @@ -187,6 +201,54 @@ def test_send_x5c_explicit_false_overrides_kwarg(self): ) assert auth_config.SEND_X5C is False + def test_certificate_subject_name_from_parameter(self): + auth_config = AgentAuthConfiguration( + auth_type=AuthTypes.certificate_subject_name, + cert_subject_name="CN=test-agent", + ) + assert auth_config.CERT_SUBJECT_NAME == "CN=test-agent" + + def test_certificate_subject_name_from_kwargs(self): + auth_config = AgentAuthConfiguration( + AUTHTYPE="CertificateSubjectName", + CERTSUBJECTNAME="CN=test-agent", + ) + assert auth_config.CERT_SUBJECT_NAME == "CN=test-agent" + assert "CERTSUBJECTNAME" not in auth_config.provider_settings + + def test_certificate_store_name_defaults_my(self): + auth_config = AgentAuthConfiguration() + assert auth_config.CERT_STORE_NAME == "My" + + def test_certificate_store_name_from_parameter(self): + auth_config = AgentAuthConfiguration(cert_store_name="Root") + assert auth_config.CERT_STORE_NAME == "Root" + + def test_certificate_store_name_from_kwargs(self): + auth_config = AgentAuthConfiguration(CERTSTORENAME="Root") + assert auth_config.CERT_STORE_NAME == "Root" + assert "CERTSTORENAME" not in auth_config.provider_settings + + def test_valid_certificate_only_defaults_true(self): + auth_config = AgentAuthConfiguration() + assert auth_config.VALID_CERTIFICATE_ONLY is True + + def test_valid_certificate_only_false_from_parameter(self): + auth_config = AgentAuthConfiguration(valid_certificate_only=False) + assert auth_config.VALID_CERTIFICATE_ONLY is False + + def test_valid_certificate_only_from_kwargs(self): + auth_config = AgentAuthConfiguration(VALIDCERTIFICATEONLY="false") + assert auth_config.VALID_CERTIFICATE_ONLY is False + assert "VALIDCERTIFICATEONLY" not in auth_config.provider_settings + + def test_valid_certificate_only_explicit_false_overrides_kwarg(self): + auth_config = AgentAuthConfiguration( + valid_certificate_only=False, + VALIDCERTIFICATEONLY="true", + ) + assert auth_config.VALID_CERTIFICATE_ONLY is False + def test_azure_region_from_parameter(self): auth_config = AgentAuthConfiguration(azure_region="westus") assert auth_config.AZURE_REGION == "westus"