AT POP for Public Client based on broker

Pop test case

msal/application.py

@@ -1204,6 +1204,7 @@ def acquire_token_silent(
             authority=None,  # See get_authorization_request_url()
             force_refresh=False,  # type: Optional[boolean]
             claims_challenge=None,
+            auth_scheme=None,
             **kwargs):
         """Acquire an access token for given account, without user interaction.
 
@@ -1224,6 +1225,12 @@ def acquire_token_silent(
             returned from the UserInfo Endpoint and/or in the ID Token and/or Access Token.
             It is a string of a JSON object which contains lists of claims being requested from these locations.
 
+        :param object auth_scheme:
+            You can provide an ``msal.auth_scheme.PopAuthScheme`` object
+            so that MSAL will get a Proof-of-Possession (POP) token for you.
+
+            New in version 1.21.0.
+
         :return:
             - A dict containing no "error" key,
               and typically contains an "access_token" key,
@@ -1232,7 +1239,7 @@ def acquire_token_silent(
         """
         result = self.acquire_token_silent_with_error(
             scopes, account, authority=authority, force_refresh=force_refresh,
-            claims_challenge=claims_challenge, **kwargs)
+            claims_challenge=claims_challenge, auth_scheme=auth_scheme, **kwargs)
         return result if result and "error" not in result else None
 
     def acquire_token_silent_with_error(
@@ -1242,6 +1249,7 @@ def acquire_token_silent_with_error(
             authority=None,  # See get_authorization_request_url()
             force_refresh=False,  # type: Optional[boolean]
             claims_challenge=None,
+            auth_scheme=None,
             **kwargs):
         """Acquire an access token for given account, without user interaction.
 
@@ -1267,6 +1275,12 @@ def acquire_token_silent_with_error(
             in the form of a claims_challenge directive in the www-authenticate header to be
             returned from the UserInfo Endpoint and/or in the ID Token and/or Access Token.
             It is a string of a JSON object which contains lists of claims being requested from these locations.
+        :param object auth_scheme:
+            You can provide an ``msal.auth_scheme.PopAuthScheme`` object
+            so that MSAL will get a Proof-of-Possession (POP) token for you.
+
+            New in version 1.21.0.
+
         :return:
             - A dict containing no "error" key,
               and typically contains an "access_token" key,
@@ -1288,6 +1302,7 @@ def acquire_token_silent_with_error(
             scopes, account, self.authority, force_refresh=force_refresh,
             claims_challenge=claims_challenge,
             correlation_id=correlation_id,
+            auth_scheme=auth_scheme,
             **kwargs)
         if result and "error" not in result:
             return result
@@ -1310,6 +1325,7 @@ def acquire_token_silent_with_error(
                 scopes, account, the_authority, force_refresh=force_refresh,
                 claims_challenge=claims_challenge,
                 correlation_id=correlation_id,
+                auth_scheme=auth_scheme,
                 **kwargs)
             if result:
                 if "error" not in result:
@@ -1333,9 +1349,10 @@ def _acquire_token_silent_from_cache_and_possibly_refresh_it(
             force_refresh=False,  # type: Optional[boolean]
             claims_challenge=None,
             correlation_id=None,
+            auth_scheme=None,
             **kwargs):
         access_token_from_cache = None
-        if not (force_refresh or claims_challenge):  # Bypass AT when desired or using claims
+        if not (force_refresh or claims_challenge or auth_scheme):  # Then attempt AT cache
             query={
                     "client_id": self.client_id,
                     "environment": authority.instance,
@@ -1373,6 +1390,8 @@ def _acquire_token_silent_from_cache_and_possibly_refresh_it(
         try:
             data = kwargs.get("data", {})
             if account and account.get("authority_type") == _AUTHORITY_TYPE_CLOUDSHELL:
+                if auth_scheme:
+                    raise ValueError("auth_scheme is not supported in Cloud Shell")
                 return self._acquire_token_by_cloud_shell(scopes, data=data)
 
             if self._enable_broker and account is not None and data.get("token_type") != "ssh-cert":
@@ -1385,10 +1404,13 @@ def _acquire_token_silent_from_cache_and_possibly_refresh_it(
                     claims=_merge_claims_challenge_and_capabilities(
                         self._client_capabilities, claims_challenge),
                     correlation_id=correlation_id,
+                    auth_scheme=auth_scheme,
                     **data)
                 if response:  # The broker provided a decisive outcome, so we use it
                     return self._process_broker_response(response, scopes, data)
 
+            if auth_scheme:
+                raise ValueError("auth_scheme is currently only available from broker")
             result = _clean_up(self._acquire_token_silent_by_finding_rt_belongs_to_me_or_my_family(
                 authority, self._decorate_scope(scopes), account,
                 refresh_reason=refresh_reason, claims_challenge=claims_challenge,
@@ -1569,7 +1591,11 @@ def acquire_token_by_refresh_token(self, refresh_token, scopes, **kwargs):
         return response
 
     def acquire_token_by_username_password(
-            self, username, password, scopes, claims_challenge=None, **kwargs):
+            self, username, password, scopes, claims_challenge=None,
+            # Note: We shouldn't need to surface enable_msa_passthrough,
+            # because this ROPC won't work with MSA account anyway.
+            auth_scheme=None,
+            **kwargs):
         """Gets a token for a given resource via user credentials.
 
         See this page for constraints of Username Password Flow.
@@ -1585,6 +1611,12 @@ def acquire_token_by_username_password(
             returned from the UserInfo Endpoint and/or in the ID Token and/or Access Token.
             It is a string of a JSON object which contains lists of claims being requested from these locations.
 
+        :param object auth_scheme:
+            You can provide an ``msal.auth_scheme.PopAuthScheme`` object
+            so that MSAL will get a Proof-of-Possession (POP) token for you.
+
+            New in version 1.21.0.
+
         :return: A dict representing the json response from AAD:
 
             - A successful response would contain "access_token" key,
@@ -1604,9 +1636,12 @@ def acquire_token_by_username_password(
                     self.authority._is_known_to_developer
                     or self._instance_discovery is False) else None,
                 claims=claims,
+                auth_scheme=auth_scheme,
                 )
             return self._process_broker_response(response, scopes, kwargs.get("data", {}))
 
+        if auth_scheme:
+            raise ValueError("auth_scheme is currently only available from broker")
         scopes = self._decorate_scope(scopes)
         telemetry_context = self._build_telemetry_context(
             self.ACQUIRE_TOKEN_BY_USERNAME_PASSWORD_ID)
@@ -1698,6 +1733,7 @@ def acquire_token_interactive(
             max_age=None,
             parent_window_handle=None,
             on_before_launching_ui=None,
+            auth_scheme=None,
             **kwargs):
         """Acquire token interactively i.e. via a local browser.
 
@@ -1773,6 +1809,12 @@ def acquire_token_interactive(
 
             New in version 1.20.0.
 
+        :param object auth_scheme:
+            You can provide an ``msal.auth_scheme.PopAuthScheme`` object
+            so that MSAL will get a Proof-of-Possession (POP) token for you.
+
+            New in version 1.21.0.
+
         :return:
             - A dict containing no "error" key,
               and typically contains an "access_token" key.
@@ -1817,11 +1859,14 @@ def acquire_token_interactive(
                 claims,
                 data,
                 on_before_launching_ui,
+                auth_scheme,
                 prompt=prompt,
                 login_hint=login_hint,
                 max_age=max_age,
                 )
 
+        if auth_scheme:
+            raise ValueError("auth_scheme is currently only available from broker")
         on_before_launching_ui(ui="browser")
         telemetry_context = self._build_telemetry_context(
             self.ACQUIRE_TOKEN_INTERACTIVE)
@@ -1854,6 +1899,7 @@ def _acquire_token_interactive_via_broker(
             claims,  # type: str
             data,  # type: dict
             on_before_launching_ui,  # type: callable
+            auth_scheme,  # type: object
             prompt=None,
             login_hint=None,  # type: Optional[str]
             max_age=None,
@@ -1877,6 +1923,7 @@ def _acquire_token_interactive_via_broker(
                     accounts[0]["local_account_id"],
                     scopes,
                     claims=claims,
+                    auth_scheme=auth_scheme,
                     **data)
                 if response and "error" not in response:
                     return self._process_broker_response(response, scopes, data)
@@ -1889,6 +1936,7 @@ def _acquire_token_interactive_via_broker(
                 claims=claims,
                 max_age=max_age,
                 enable_msa_pt=enable_msa_passthrough,
+                auth_scheme=auth_scheme,
                 **data)
             is_wrong_account = bool(
                 # _signin_silently() only gets tokens for default account,
@@ -1931,6 +1979,7 @@ def _acquire_token_interactive_via_broker(
             claims=claims,
             max_age=max_age,
             enable_msa_pt=enable_msa_passthrough,
+            auth_scheme=auth_scheme,
             **data)
         return self._process_broker_response(response, scopes, data)
 

msal/auth_scheme.py

@@ -0,0 +1,28 @@
+try:
+    from urllib.parse import urlparse
+except ImportError:  # Fall back to Python 2
+    from urlparse import urlparse
+
+# We may support more auth schemes in the future
+class PopAuthScheme(object):
+    # Internal design: https://identitydivision.visualstudio.com/DevEx/_git/AuthLibrariesApiReview?path=/PoPTokensProtocol/PopTokensProtocol.md
+    def __init__(self, http_method=None, url=None, nonce=None):
+        """Create an auth scheme which is needed to obtain a Proof-of-Possession token.
+
+        :param str http_method:
+            Its value is an uppercase http verb, such as "GET" and "POST".
+        :param str url:
+            The url to be signed.
+        :param str nonce:
+            The nonce came from resource's challenge.
+        """
+        if not (http_method and url and nonce):
+            # In the future, we may also support accepting an http_response as input
+            raise ValueError("All http_method, url and nonce are required parameters")
+        if http_method.upper() != http_method:
+            raise ValueError("http_method must be uppercase, according to "
+                "https://datatracker.ietf.org/doc/html/draft-ietf-oauth-signed-http-request-03#section-3")
+        self._http_method = http_method
+        self._url = urlparse(url)
+        self._nonce = nonce
+

msal/broker.py

@@ -92,13 +92,17 @@ def _convert_result(result, client_id, expected_token_type=None):  # Mimic an on
     assert account, "Account is expected to be always available"
     # Note: There are more account attribute getters available in pymsalruntime 0.13+
     return_value = {k: v for k, v in {
-        "access_token": result.get_access_token(),
+        "access_token":
+            result.get_authorization_header()  # It returns "pop SignedHttpRequest"
+                .split()[1]
+            if result.is_pop_authorization() else result.get_access_token(),
         "expires_in": result.get_access_token_expiry_time() - int(time.time()),  # Convert epoch to count-down
         "id_token": result.get_raw_id_token(),  # New in pymsalruntime 0.8.1
         "id_token_claims": id_token_claims,
         "client_info": account.get_client_info(),
         "_account_id": account.get_account_id(),
-		"token_type": expected_token_type or "Bearer",  # Workaround its absence from broker
+		"token_type": "pop" if result.is_pop_authorization() else (
+            expected_token_type or "bearer"),  # Workaround "ssh-cert"'s absence from broker
         }.items() if v}
     likely_a_cert = return_value["access_token"].startswith("AAAA")  # Empirical observation
     if return_value["token_type"].lower() == "ssh-cert" and not likely_a_cert:
@@ -121,11 +125,16 @@ def _enable_msa_pt(params):
 def _signin_silently(
         authority, client_id, scopes, correlation_id=None, claims=None,
         enable_msa_pt=False,
+        auth_scheme=None,
         **kwargs):
     params = pymsalruntime.MSALRuntimeAuthParameters(client_id, authority)
     params.set_requested_scopes(scopes)
     if claims:
         params.set_decoded_claims(claims)
+    if auth_scheme:
+        params.set_pop_params(
+            auth_scheme._http_method, auth_scheme._url.netloc, auth_scheme._url.path,
+            auth_scheme._nonce)
     callback_data = _CallbackData()
     for k, v in kwargs.items():  # This can be used to support domain_hint, max_age, etc.
         if v is not None:
@@ -149,6 +158,7 @@ def _signin_interactively(
         claims=None,
         correlation_id=None,
         enable_msa_pt=False,
+        auth_scheme=None,
         **kwargs):
     params = pymsalruntime.MSALRuntimeAuthParameters(client_id, authority)
     params.set_requested_scopes(scopes)
@@ -171,6 +181,10 @@ def _signin_interactively(
         params.set_additional_parameter("msal_gui_thread", "true")  # Since pymsalruntime 0.8.1
     if enable_msa_pt:
         _enable_msa_pt(params)
+    if auth_scheme:
+        params.set_pop_params(
+            auth_scheme._http_method, auth_scheme._url.netloc, auth_scheme._url.path,
+            auth_scheme._nonce)
     for k, v in kwargs.items():  # This can be used to support domain_hint, max_age, etc.
         if v is not None:
             params.set_additional_parameter(k, str(v))
@@ -190,6 +204,7 @@ def _signin_interactively(
 
 def _acquire_token_silently(
         authority, client_id, account_id, scopes, claims=None, correlation_id=None,
+        auth_scheme=None,
         **kwargs):
     # For MSA PT scenario where you use the /organizations, yes,
     # acquireTokenSilently is expected to fail.  - Sam Wilson
@@ -203,6 +218,10 @@ def _acquire_token_silently(
     params.set_requested_scopes(scopes)
     if claims:
         params.set_decoded_claims(claims)
+    if auth_scheme:
+        params.set_pop_params(
+            auth_scheme._http_method, auth_scheme._url.netloc, auth_scheme._url.path,
+            auth_scheme._nonce)
     for k, v in kwargs.items():  # This can be used to support domain_hint, max_age, etc.
         if v is not None:
             params.set_additional_parameter(k, str(v))

tests/msaltest.py

@@ -1,4 +1,12 @@
 import getpass, logging, pprint, sys, msal
+from msal.auth_scheme import PopAuthScheme
+
+
+placeholder_auth_scheme = PopAuthScheme(
+    http_method="GET",
+    url="https://example.com/endpoint",
+    nonce="placeholder",
+    )
 
 
 AZURE_CLI = "04b07795-8ddb-461a-bbee-02f9e1bf7b46"
@@ -63,6 +71,7 @@ def acquire_token_silent(app):
             _input_scopes(),
             account=account,
             force_refresh=_input_boolean("Bypass MSAL Python's token cache?"),
+            auth_scheme=placeholder_auth_scheme if app._enable_broker else None,
             ))
 
 def _acquire_token_interactive(app, scopes, data=None):
@@ -87,7 +96,9 @@ def _acquire_token_interactive(app, scopes, data=None):
         enable_msa_passthrough=app.client_id in [  # Apps are expected to set this right
             AZURE_CLI, VISUAL_STUDIO,
             ],  # Here this test app mimics the setting for some known MSA-PT apps
-        prompt=prompt, login_hint=login_hint, data=data or {})
+        prompt=prompt, login_hint=login_hint, data=data or {},
+        auth_scheme=placeholder_auth_scheme if app._enable_broker else None,
+        )
     if login_hint and "id_token_claims" in result:
         signed_in_user = result.get("id_token_claims", {}).get("preferred_username")
         if signed_in_user != login_hint: