[Corehttp] Support no auth in auth policy (Azure#41274)

Enable per-operation auth_flow configuration, and also allow an empty
auth_flow to indicate no auth.

Signed-off-by: Paul Van Eck <[email protected]>

Author: pvaneck

sdk/core/corehttp/corehttp/runtime/policies/_authentication.py

@@ -110,6 +110,9 @@ def on_request(
         :keyword auth_flows: A list of authentication flows to use for the credential.
         :paramtype auth_flows: list[dict[str, Union[str, list[dict[str, str]]]]]
         """
+        # If auth_flows is an empty list, we should not attempt to authorize the request.
+        if auth_flows is not None and len(auth_flows) == 0:
+            return
         self._enforce_https(request)
 
         if self._token is None or self._need_new_token:
@@ -142,7 +145,9 @@ def send(self, request: PipelineRequest[HTTPRequestType]) -> PipelineResponse[HT
         :return: The pipeline response object
         :rtype: ~corehttp.runtime.pipeline.PipelineResponse
         """
-        self.on_request(request, auth_flows=self._auth_flows)
+        op_auth_flows = request.context.options.pop("auth_flows", None)
+        auth_flows = op_auth_flows if op_auth_flows is not None else self._auth_flows
+        self.on_request(request, auth_flows=auth_flows)
         try:
             response = self.next.send(request)
         except Exception:

sdk/core/corehttp/corehttp/runtime/policies/_authentication_async.py

@@ -68,6 +68,9 @@ async def on_request(
         :paramtype auth_flows: list[dict[str, Union[str, list[dict[str, str]]]]]
         :raises: :class:`~corehttp.exceptions.ServiceRequestError`
         """
+        # If auth_flows is an empty list, we should not attempt to authorize the request.
+        if auth_flows is not None and len(auth_flows) == 0:
+            return
         _BearerTokenCredentialPolicyBase._enforce_https(request)  # pylint:disable=protected-access
 
         if self._token is None or self._need_new_token:
@@ -107,7 +110,9 @@ async def send(
         :return: The pipeline response object
         :rtype: ~corehttp.runtime.pipeline.PipelineResponse
         """
-        await await_result(self.on_request, request, auth_flows=self._auth_flows)
+        op_auth_flows = request.context.options.pop("auth_flows", None)
+        auth_flows = op_auth_flows if op_auth_flows is not None else self._auth_flows
+        await await_result(self.on_request, request, auth_flows=auth_flows)
         try:
             response = await self.next.send(request)
         except Exception:

sdk/core/corehttp/tests/async_tests/test_authentication_async.py

@@ -323,14 +323,71 @@ async def test_need_new_token():
 
 @pytest.mark.asyncio
 async def test_send_with_auth_flows():
-    auth_flows = [{"type": "flow1"}, {"type": "flow2"}]
+    auth_flows = [
+        {
+            "tokenUrl": "https://login.microsoftonline.com/common/oauth2/token",
+            "type": "client_credentials",
+            "scopes": [
+                {"value": "https://test.microsoft.com/.default"},
+            ],
+        }
+    ]
     credential = Mock(
         spec_set=["get_token_info"],
         get_token_info=Mock(return_value=get_completed_future(AccessTokenInfo("***", int(time.time()) + 3600))),
     )
-    policy = AsyncBearerTokenCredentialPolicy(credential, "scope", auth_flows=auth_flows)
+    policy = AsyncBearerTokenCredentialPolicy(credential, "https://test.microsoft.com/.default", auth_flows=auth_flows)
     transport = Mock(send=Mock(return_value=get_completed_future(Mock(status_code=200))))
 
     pipeline = AsyncPipeline(transport=transport, policies=[policy])
     await pipeline.run(HttpRequest("GET", "https://localhost"))
-    policy._credential.get_token_info.assert_called_with("scope", options={"auth_flows": auth_flows})
+    policy._credential.get_token_info.assert_called_with(
+        "https://test.microsoft.com/.default", options={"auth_flows": auth_flows}
+    )
+
+
+@pytest.mark.asyncio
+async def test_disable_authorization_header():
+    """Tests that we can disable the Authorization header in the request"""
+    credential = Mock(
+        spec_set=["get_token_info"],
+        get_token_info=Mock(return_value=get_completed_future(AccessTokenInfo("***", int(time.time()) + 3600))),
+    )
+    policy = AsyncBearerTokenCredentialPolicy(credential, "scope", auth_flows={"foo": "bar"})
+    transport = Mock(send=Mock(return_value=get_completed_future(Mock(status_code=200))))
+
+    pipeline = AsyncPipeline(transport=transport, policies=[policy])
+    request = HttpRequest("GET", "https://localhost")
+    await pipeline.run(request, auth_flows=[])
+    assert "Authorization" not in request.headers
+
+
+@pytest.mark.asyncio
+async def test_auth_flow_operation_override():
+    """Tests that the operation level auth_flow is passed to the credential's get_token_info method."""
+    auth_flows = [
+        {
+            "tokenUrl": "https://login.microsoftonline.com/common/oauth2/token",
+            "type": "client_credentials",
+            "scopes": [{"value": "https://graph.microsoft.com/.default"}],
+        }
+    ]
+    credential = Mock(
+        spec_set=["get_token_info"],
+        get_token_info=Mock(return_value=get_completed_future(AccessTokenInfo("***", int(time.time()) + 3600))),
+    )
+    policy = AsyncBearerTokenCredentialPolicy(credential, "https://graph.microsoft.com/.default", auth_flows=auth_flows)
+    transport = Mock(send=Mock(return_value=get_completed_future(Mock(status_code=200))))
+
+    pipeline = AsyncPipeline(transport=transport, policies=[policy])
+    op_auth_flow = [
+        {
+            "tokenUrl": "https://login.microsoftonline.com/common/oauth2/token",
+            "type": "client_credentials",
+            "scopes": [{"value": "https://foo.microsoft.com/.default"}],
+        }
+    ]
+    await pipeline.run(HttpRequest("GET", "https://localhost"), auth_flows=op_auth_flow)
+    policy._credential.get_token_info.assert_called_with(
+        "https://graph.microsoft.com/.default", options={"auth_flows": op_auth_flow}
+    )

sdk/core/corehttp/tests/test_authentication.py

@@ -418,14 +418,70 @@ def test_need_new_token():
 
 
 def test_send_with_auth_flows():
-    auth_flows = [{"type": "flow1"}, {"type": "flow2"}]
+    auth_flows = [
+        {
+            "tokenUrl": "https://login.microsoftonline.com/common/oauth2/token",
+            "type": "client_credentials",
+            "scopes": [
+                {"value": "https://test.microsoft.com/.default"},
+            ],
+        }
+    ]
     credential = Mock(
         spec_set=["get_token_info"],
         get_token_info=Mock(return_value=AccessTokenInfo("***", int(time.time()) + 3600)),
     )
-    policy = BearerTokenCredentialPolicy(credential, "scope", auth_flows=auth_flows)
+    policy = BearerTokenCredentialPolicy(credential, "https://test.microsoft.com/.default", auth_flows=auth_flows)
     transport = Mock(send=Mock(return_value=Mock(status_code=200)))
 
     pipeline = Pipeline(transport=transport, policies=[policy])
     pipeline.run(HttpRequest("GET", "https://localhost"))
-    policy._credential.get_token_info.assert_called_with("scope", options={"auth_flows": auth_flows})
+    policy._credential.get_token_info.assert_called_with(
+        "https://test.microsoft.com/.default", options={"auth_flows": auth_flows}
+    )
+
+
+def test_disable_authorization_header():
+    """Tests that we can disable the Authorization header in the request"""
+    credential = Mock(
+        spec_set=["get_token_info"],
+        get_token_info=Mock(return_value=AccessTokenInfo("***", int(time.time()) + 3600)),
+    )
+    policy = BearerTokenCredentialPolicy(credential, "scope", auth_flows={"foo": "bar"})
+    transport = Mock(send=Mock(return_value=Mock(status_code=200)))
+
+    pipeline = Pipeline(transport=transport, policies=[policy])
+    request = HttpRequest("GET", "https://localhost")
+    pipeline.run(request, auth_flows=[])
+    assert "Authorization" not in request.headers
+
+
+def test_auth_flow_operation_override():
+    """Tests that the operation level auth_flow is passed to the credential's get_token_info method."""
+    auth_flows = [
+        {
+            "tokenUrl": "https://login.microsoftonline.com/common/oauth2/token",
+            "type": "client_credentials",
+            "scopes": [{"value": "https://graph.microsoft.com/.default"}],
+        }
+    ]
+    credential = Mock(
+        spec_set=["get_token_info"],
+        get_token_info=Mock(return_value=AccessTokenInfo("***", int(time.time()) + 3600)),
+    )
+    policy = BearerTokenCredentialPolicy(credential, "https://graph.microsoft.com/.default", auth_flows=auth_flows)
+    transport = Mock(send=Mock(return_value=Mock(status_code=200)))
+
+    op_auth_flow = [
+        {
+            "tokenUrl": "https://login.microsoftonline.com/common/oauth2/token",
+            "type": "client_credentials",
+            "scopes": [{"value": "https://foo.microsoft.com/.default"}],
+        }
+    ]
+
+    pipeline = Pipeline(transport=transport, policies=[policy])
+    pipeline.run(HttpRequest("GET", "https://localhost"), auth_flows=op_auth_flow)
+    policy._credential.get_token_info.assert_called_with(
+        "https://graph.microsoft.com/.default", options={"auth_flows": op_auth_flow}
+    )