PoC: req_ds_cnf

Author: rayluo

msal/application.py

@@ -1,10 +1,12 @@
+import base64
 import functools
 import json
 import time
 import logging
 import sys
 import warnings
 from threading import Lock
+from typing import Optional  # Needed in Python 3.7 & 3.8
 import os
 
 from .oauth2cli import Client, JwtAssertionCreator
@@ -164,6 +166,17 @@ def _preferred_browser():
     return None
 
 
+def _build_req_cnf(jwk:dict, remove_padding:bool = False) -> str:
+    """req_cnf usually requires base64url encoding.
+
+    https://datatracker.ietf.org/doc/html/draft-ietf-oauth-pop-key-distribution-07#section-4.2.1
+    https://learn.microsoft.com/en-us/openspecs/windows_protocols/ms-rdpbcgr/e967ebeb-9e9f-443e-857a-5208802943c2
+    """
+    raw = json.dumps(jwk)
+    encoded = base64.urlsafe_b64encode(raw.encode('utf-8')).decode('utf-8')
+    return encoded.rstrip('=') if remove_padding else encoded
+
+
 class _ClientWithCcsRoutingInfo(Client):
 
     def initiate_auth_code_flow(self, **kwargs):
@@ -2344,6 +2357,9 @@ def _acquire_token_for_client(
         scopes,
         refresh_reason,
         claims_challenge=None,
+        *,
+        delegation_constraints: Optional[list] = None,
+        req_ds_cnf: Optional[dict] = None,
         **kwargs
     ):
         if self.authority.tenant.lower() in ["common", "organizations"]:
@@ -2360,9 +2376,15 @@ def _acquire_token_for_client(
             headers=telemetry_context.generate_headers(),
             data=dict(
                 kwargs.pop("data", {}),
+                req_ds_cnf=_build_req_cnf(req_ds_cnf) if req_ds_cnf else None,
                 claims=_merge_claims_challenge_and_capabilities(
                     self._client_capabilities, claims_challenge)),
             **kwargs)
+        if (
+            req_ds_cnf
+            and not response.get("error") and not response.get("xms_ds_nonce")
+        ):
+            raise ValueError("Your app shall opt in to xms_ds_cnf claim first")
         telemetry_context.update_telemetry(response)
         return response
 

tests/test_e2e.py

@@ -29,6 +29,7 @@
 from tests.http_client import MinimalHttpClient, MinimalResponse
 from msal.oauth2cli import AuthCodeReceiver
 from msal.oauth2cli.oidc import decode_part
+from msal.application import _build_req_cnf
 
 try:
     import pymsalruntime
@@ -791,12 +792,12 @@ def test_user_account(self):
         self._test_user_account()
 
 
-def _data_for_pop(key):
-    raw_req_cnf = json.dumps({"kid": key, "xms_ksl": "sw"})
+def _data_for_pop(key_id):
     return {  # Sampled from Azure CLI's plugin connectedk8s
         'token_type': 'pop',
-        'key_id': key,
-        "req_cnf": base64.urlsafe_b64encode(raw_req_cnf.encode('utf-8')).decode('utf-8').rstrip('='),
+        'key_id': key_id,
+        "req_cnf": _build_req_cnf(
+            {"kid": key_id, "xms_ksl": "sw"}, remove_padding=True),
             # Note: Sending raw_req_cnf without base64 encoding would result in an http 500 error
     }  # See also https://github.com/Azure/azure-cli-extensions/blob/main/src/connectedk8s/azext_connectedk8s/_clientproxyutils.py#L86-L92
 
@@ -817,6 +818,38 @@ def test_user_account(self):
         self._test_user_account()
 
 
+class CdtTestCase(LabBasedTestCase):
+    _JWK1 = {"kty":"RSA", "n":"2tNr73xwcj6lH7bqRZrFzgSLj7OeLfbn8216uOMDHuaZ6TEUBDN8Uz0ve8jAlKsP9CQFCSVoSNovdE-fs7c15MxEGHjDcNKLWonznximj8pDGZQjVdfK-7mG6P6z-lgVcLuYu5JcWU_PeEqIKg5llOaz-qeQ4LEDS4T1D2qWRGpAra4rJX1-kmrWmX_XIamq30C9EIO0gGuT4rc2hJBWQ-4-FnE1NXmy125wfT3NdotAJGq5lMIfhjfglDbJCwhc8Oe17ORjO3FsB5CLuBRpYmP7Nzn66lRY3Fe11Xz8AEBl3anKFSJcTvlMnFtu3EpD-eiaHfTgRBU7CztGQqVbiQ", "e":"AQAB"}
+    def test_service_principal(self):
+        """
+        app = get_lab_app(
+            authority="https://login.microsoftonline.com/microsoft.onmicrosoft.com"
+            "?dc=ESTS-PUB-JPELR1-AZ1-FD000-TEST1",
+            )
+        """
+        app = msal.ConfidentialClientApplication(
+            os.getenv("RAY_APP_CLIENT_ID"),
+            client_credential=os.getenv("RAY_APP_CLIENT_SECRET"),
+            authority="https://login.microsoftonline.com/msidlab4.onmicrosoft.com"
+            "?dc=ESTS-PUB-JPELR1-AZ1-FD000-TEST1",  # Accessible within AzVPN
+        )
+        from http.client import HTTPConnection
+        HTTPConnection.debuglevel = 1
+        result = app.acquire_token_for_client(
+            [f"{app.client_id}/.default"],
+            delegation_constraints=[
+                {"typ": "usr", "a": "C", "target": ["constraint1", "constraint4"]},
+                {"typ": "app", "a": "R", "target": ["constraint2", "constraint5"]},
+                {"typ": "subscription", "a": "U", "target": ["constraint3"]},
+            ],
+            req_ds_cnf=self._JWK1,
+        )
+        self.assertIsNotNone(result.get("access_token"), "Encountered {}: {}".format(
+            result.get("error"), result.get("error_description")))
+        print("Test case result:", result)
+        self.assertIsNotNone(result.get("xms_ds_nonce"))
+
+
 class WorldWideTestCase(LabBasedTestCase):
 
     def test_aad_managed_user(self):  # Pure cloud