[Identity] Use form_post for InteractiveBrowserCredential (#46598)

Signed-off-by: Paul Van Eck <paulvaneck@microsoft.com>

Author: pvaneck

sdk/identity/azure-identity/azure/identity/_credentials/browser.py

@@ -115,6 +115,7 @@ def _request_token(self, *scopes: str, **kwargs) -> Dict:
             prompt="select_account",
             claims_challenge=claims,
             login_hint=self._login_hint,
+            response_mode="form_post",
         )
         if "auth_uri" not in flow:
             raise CredentialUnavailableError("Failed to begin authentication flow")

sdk/identity/azure-identity/azure/identity/_internal/auth_code_redirect_handler.py

@@ -20,12 +20,20 @@ def do_GET(self):
 
         query = self.path.split("?", 1)[-1]
         parsed = parse_qs(query, keep_blank_values=True)
-        self.server.query_params = {k: v[0] if isinstance(v, list) and len(v) == 1 else v for k, v in parsed.items()}
+        self.server.auth_response = {k: v[0] if isinstance(v, list) and len(v) == 1 else v for k, v in parsed.items()}
+        self._send_success_response()
 
+    def do_POST(self):
+        content_length = int(self.headers.get("Content-Length", 0))
+        body = self.rfile.read(content_length).decode("utf-8")
+        parsed = parse_qs(body, keep_blank_values=True)
+        self.server.auth_response = {k: v[0] if isinstance(v, list) and len(v) == 1 else v for k, v in parsed.items()}
+        self._send_success_response()
+
+    def _send_success_response(self):
         self.send_response(200)
         self.send_header("Content-Type", "text/html")
         self.end_headers()
-
         self.wfile.write(b"Authentication complete. You can close this window.")
 
     def log_message(self, format, *args):  # pylint: disable=redefined-builtin
@@ -35,14 +43,13 @@ def log_message(self, format, *args):  # pylint: disable=redefined-builtin
 class AuthCodeRedirectServer(HTTPServer):
     """HTTP server that listens for the redirect request following an authorization code authentication"""
 
-    query_params: Mapping[str, Any] = {}
-
     def __init__(self, hostname: str, port: int, timeout: int) -> None:
         HTTPServer.__init__(self, (hostname, port), AuthCodeRedirectHandler)
         self.timeout = timeout
+        self.auth_response: Mapping[str, Any] = {}
 
     def wait_for_redirect(self) -> Mapping[str, Any]:
-        while not self.query_params:
+        while not self.auth_response:
             try:
                 self.handle_request()
             except (IOError, ValueError):
@@ -53,7 +60,7 @@ def wait_for_redirect(self) -> Mapping[str, Any]:
         self.server_close()
 
         # if we timed out, this returns an empty dict
-        return self.query_params
+        return self.auth_response
 
     def handle_timeout(self):
         """Break the request-handling loop by tearing down the server"""

sdk/identity/azure-identity/tests/test_browser_credential.py

@@ -167,7 +167,40 @@ def test_redirect_server(get_token_method):
     response = urllib.request.urlopen(url)  # nosec
 
     assert response.code == 200
-    assert server.query_params[expected_param] == expected_value
+    assert server.auth_response[expected_param] == expected_value
+
+
+@pytest.mark.parametrize("get_token_method", GET_TOKEN_METHODS)
+def test_redirect_server_post(get_token_method):
+    """The redirect server should handle POST requests with form-encoded bodies (form_post response mode)"""
+
+    server = None
+    hostname = "127.0.0.1"
+    for _ in range(4):
+        try:
+            port = random.randint(1024, 65535)
+            server = AuthCodeRedirectServer(hostname, port, timeout=10)
+            break
+        except socket.error:
+            continue
+
+    assert server, "failed to start redirect server"
+
+    expected_param = "code"
+    expected_value = "test-auth-code"
+
+    thread = threading.Thread(target=server.wait_for_redirect)
+    thread.daemon = True
+    thread.start()
+
+    # send a POST request with form-encoded body, simulating form_post response mode
+    url = "http://{}:{}".format(hostname, port)
+    data = urllib.parse.urlencode({expected_param: expected_value}).encode("utf-8")
+    request = urllib.request.Request(url, data=data, headers={"Content-Type": "application/x-www-form-urlencoded"})
+    response = urllib.request.urlopen(request)  # nosec
+
+    assert response.code == 200
+    assert server.auth_response[expected_param] == expected_value
 
 
 @pytest.mark.parametrize("get_token_method", GET_TOKEN_METHODS)