|
| 1 | +"""Helpers for mTLS Proof-of-Possession (PoP). |
| 2 | +
|
| 3 | +This module owns two concerns for the "SN/I certificate over mTLS PoP" feature: |
| 4 | +
|
| 5 | +1. The endpoint transform + sovereign guardrail: mapping a tenanted ``login.*`` |
| 6 | + authority to its ``mtlsauth.*`` counterpart (global or regional), and |
| 7 | + rejecting clouds/hosts where mTLS PoP is not (yet) available. |
| 8 | +2. The mTLS transport: a ``requests`` session whose HTTPS connections present a |
| 9 | + client certificate for the mutual-TLS handshake to the token endpoint. |
| 10 | +
|
| 11 | +The wire contract and host mapping mirror the shipped MSAL.NET |
| 12 | +``RegionAndMtlsDiscoveryProvider`` so MSAL Python stays cross-SDK consistent. |
| 13 | +""" |
| 14 | +import logging |
| 15 | +try: |
| 16 | + from urllib.parse import urlparse, urlunparse |
| 17 | +except ImportError: # Python 2 |
| 18 | + from urlparse import urlparse, urlunparse |
| 19 | + |
| 20 | + |
| 21 | +logger = logging.getLogger(__name__) |
| 22 | + |
| 23 | +_MTLS_POP_DOC_LINK = "https://aka.ms/msal-net-pop" |
| 24 | + |
| 25 | +# The global public mTLS host. The four public "login.*" hosts all normalize to |
| 26 | +# this single global endpoint (ESTSR provides regional failover), matching |
| 27 | +# MSAL.NET's PublicEnvForRegionalMtlsAuth. |
| 28 | +_PUBLIC_MTLS_HOST = "mtlsauth.microsoft.com" |
| 29 | + |
| 30 | +# Public-cloud login hosts that normalize to the single global mTLS host above. |
| 31 | +_PUBLIC_CLOUD_LOGIN_HOSTS = frozenset([ |
| 32 | + "login.microsoftonline.com", |
| 33 | + "login.microsoft.com", |
| 34 | + "login.windows.net", |
| 35 | + "sts.windows.net", |
| 36 | + ]) |
| 37 | + |
| 38 | +# ───────────────────────────────────────────────────────────────────────────── |
| 39 | +# SOVEREIGN GUARDRAIL - single override point for mTLS cloud availability. |
| 40 | +# |
| 41 | +# mTLS PoP is currently rejected for the deprecated sovereign login hosts below |
| 42 | +# and for any non-"login." host. ``mtlsauth.*`` is rolling out across clouds |
| 43 | +# (Azure Government / AGC: available; Bleu / Delos: TBD). To enable a cloud, |
| 44 | +# remove its entry here (and, if needed, relax the non-"login." host check in |
| 45 | +# ``mtls_pop_host``). This is the ONLY place cloud eligibility is enforced, so |
| 46 | +# do not scatter equivalent checks elsewhere in the codebase. |
| 47 | +# ───────────────────────────────────────────────────────────────────────────── |
| 48 | +_MTLS_POP_UNSUPPORTED_HOSTS = { |
| 49 | + "login.usgovcloudapi.net": |
| 50 | + "login.usgovcloudapi.net is not supported for mTLS PoP, " |
| 51 | + "please use login.microsoftonline.us", |
| 52 | + "login.chinacloudapi.cn": |
| 53 | + "login.chinacloudapi.cn is not supported for mTLS PoP, " |
| 54 | + "please use login.partner.microsoftonline.cn", |
| 55 | + } |
| 56 | + |
| 57 | + |
| 58 | +def mtls_pop_host(instance, region=None): |
| 59 | + """Return the ``mtlsauth.*`` host for a given ``login.*`` authority instance. |
| 60 | +
|
| 61 | + :param str instance: The authority host, e.g. ``login.microsoftonline.com``. |
| 62 | + :param region: Optional region, e.g. ``westus3``. When provided, a regional |
| 63 | + mTLS host ``{region}.mtlsauth...`` is returned; otherwise the global one. |
| 64 | + :raises ValueError: When mTLS PoP is not supported for the given host |
| 65 | + (the sovereign guardrail above, or a non-``login.`` host). |
| 66 | + """ |
| 67 | + instance = instance.lower() |
| 68 | + if instance in _MTLS_POP_UNSUPPORTED_HOSTS: # Sovereign guardrail |
| 69 | + raise ValueError(_MTLS_POP_UNSUPPORTED_HOSTS[instance]) |
| 70 | + if instance in _PUBLIC_CLOUD_LOGIN_HOSTS: |
| 71 | + # Known public aliases (incl. legacy login.windows.net / sts.windows.net) |
| 72 | + # all normalize to the single global mTLS host. |
| 73 | + base = _PUBLIC_MTLS_HOST |
| 74 | + elif instance.startswith("login."): # e.g. login.microsoftonline.us |
| 75 | + base = "mtlsauth" + instance[len("login"):] # -> mtlsauth.microsoftonline.us |
| 76 | + else: |
| 77 | + raise ValueError( |
| 78 | + "mTLS PoP is only supported for hosts that start with 'login.'. " |
| 79 | + "The provided authority host ({}) does not meet this requirement. " |
| 80 | + "See {} for details.".format(instance, _MTLS_POP_DOC_LINK)) |
| 81 | + return "{}.{}".format(region, base) if region else base |
| 82 | + |
| 83 | + |
| 84 | +def transform_token_endpoint(token_endpoint, instance, region=None): |
| 85 | + """Return ``token_endpoint`` with its host swapped for the mTLS host. |
| 86 | +
|
| 87 | + The path/tenant and everything else are preserved; only the network host is |
| 88 | + rewritten (e.g. ``https://login.microsoftonline.com/{tenant}/oauth2/v2.0/token`` |
| 89 | + -> ``https://mtlsauth.microsoft.com/{tenant}/oauth2/v2.0/token``). |
| 90 | + """ |
| 91 | + parsed = urlparse(token_endpoint) |
| 92 | + host = mtls_pop_host(instance, region) |
| 93 | + # Preserve a non-default port if the original endpoint carried one (tests). |
| 94 | + netloc = "{}:{}".format(host, parsed.port) if parsed.port else host |
| 95 | + return urlunparse(parsed._replace(netloc=netloc)) |
| 96 | + |
| 97 | + |
| 98 | +class _MtlsHttpClient(object): |
| 99 | + """A minimal http client (``post``/``get``/``close``) whose HTTPS |
| 100 | + connections present a client certificate for mutual-TLS. |
| 101 | +
|
| 102 | + MSAL owns this transport. A caller's plain custom ``http_client`` cannot |
| 103 | + perform the mTLS handshake, which is why requesting mTLS PoP with a |
| 104 | + non-mTLS-capable custom transport fails fast (see application.py). |
| 105 | +
|
| 106 | + The ``ssl.SSLContext`` (and its temp key file) is built lazily on first use, |
| 107 | + so Bearer-only certificate apps never pay the cost nor touch the disk. |
| 108 | + """ |
| 109 | + def __init__(self, cert_pem, key_pem, *, |
| 110 | + verify=True, proxies=None, timeout=None): |
| 111 | + # cert_pem / key_pem are PEM-encoded bytes. key_pem must be an |
| 112 | + # unencrypted private key (the caller normalizes it). |
| 113 | + self._cert_pem = cert_pem |
| 114 | + self._key_pem = key_pem |
| 115 | + self._verify = verify |
| 116 | + self._proxies = proxies |
| 117 | + self._timeout = timeout |
| 118 | + self._session = None |
| 119 | + |
| 120 | + def _ensure_session(self): |
| 121 | + if self._session is None: |
| 122 | + import requests # Lazy import, same as the rest of MSAL |
| 123 | + session = requests.Session() |
| 124 | + session.verify = self._verify |
| 125 | + if self._proxies: |
| 126 | + session.proxies = self._proxies |
| 127 | + adapter = _make_mtls_adapter( |
| 128 | + _create_ssl_context(self._cert_pem, self._key_pem)) |
| 129 | + session.mount("https://", adapter) |
| 130 | + self._session = session |
| 131 | + return self._session |
| 132 | + |
| 133 | + def post(self, url, **kwargs): |
| 134 | + if self._timeout is not None: |
| 135 | + kwargs.setdefault("timeout", self._timeout) |
| 136 | + return self._ensure_session().post(url, **kwargs) |
| 137 | + |
| 138 | + def get(self, url, **kwargs): |
| 139 | + if self._timeout is not None: |
| 140 | + kwargs.setdefault("timeout", self._timeout) |
| 141 | + return self._ensure_session().get(url, **kwargs) |
| 142 | + |
| 143 | + def close(self): |
| 144 | + if self._session is not None: |
| 145 | + self._session.close() |
| 146 | + |
| 147 | + |
| 148 | +def _make_mtls_adapter(ssl_context): |
| 149 | + """Return a ``requests`` HTTPAdapter that injects ``ssl_context`` (with the |
| 150 | + client certificate loaded) into every connection pool it creates.""" |
| 151 | + from requests.adapters import HTTPAdapter # Lazy import |
| 152 | + |
| 153 | + class _MtlsHTTPAdapter(HTTPAdapter): |
| 154 | + def __init__(self, ssl_context): |
| 155 | + self._ssl_context = ssl_context |
| 156 | + super(_MtlsHTTPAdapter, self).__init__(max_retries=1) |
| 157 | + |
| 158 | + def init_poolmanager(self, *args, **kwargs): |
| 159 | + kwargs["ssl_context"] = self._ssl_context |
| 160 | + return super(_MtlsHTTPAdapter, self).init_poolmanager(*args, **kwargs) |
| 161 | + |
| 162 | + def proxy_manager_for(self, *args, **kwargs): |
| 163 | + kwargs["ssl_context"] = self._ssl_context |
| 164 | + return super(_MtlsHTTPAdapter, self).proxy_manager_for(*args, **kwargs) |
| 165 | + |
| 166 | + return _MtlsHTTPAdapter(ssl_context) |
| 167 | + |
| 168 | + |
| 169 | +def _create_ssl_context(cert_pem, key_pem): |
| 170 | + """Build a client ``ssl.SSLContext`` that presents ``cert_pem``/``key_pem``. |
| 171 | +
|
| 172 | + ``ssl.SSLContext.load_cert_chain`` requires a file path, but our key is |
| 173 | + in memory. We write a ``0600`` temp PEM (mkstemp defaults to owner-only), |
| 174 | + load it, then unlink it immediately - the context keeps the material in |
| 175 | + memory, so nothing sensitive lingers on disk. |
| 176 | + """ |
| 177 | + import ssl |
| 178 | + import os |
| 179 | + import tempfile |
| 180 | + context = ssl.create_default_context() # Verifies the server (ESTS) as usual |
| 181 | + fd, path = tempfile.mkstemp(suffix=".pem") # Owner-only (0600) by default |
| 182 | + try: |
| 183 | + # os.fdopen() takes ownership of fd and guarantees it is closed even if |
| 184 | + # writing raises; file.write() writes every byte (os.write() may perform |
| 185 | + # a short write). Closing the fd also lets Windows unlink the file below. |
| 186 | + with os.fdopen(fd, "wb") as f: |
| 187 | + f.write(key_pem + b"\n" + cert_pem) |
| 188 | + context.load_cert_chain(path) # Loads our client cert+key into memory |
| 189 | + finally: |
| 190 | + try: |
| 191 | + os.remove(path) # Unlink immediately; minimal disk exposure |
| 192 | + except OSError: # pragma: no cover |
| 193 | + logger.warning("Unable to remove temporary mTLS key file") |
| 194 | + return context |
| 195 | + |
0 commit comments