From b098c65eb0a51ea22e08194ae06f5b33d926e8b6 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Sun, 1 Feb 2026 01:13:56 +0000 Subject: [PATCH 1/4] Initial plan From 7f24579e795e07961e3af23188a67ddcf9bc4d00 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Sun, 1 Feb 2026 01:18:14 +0000 Subject: [PATCH 2/4] Fix JWT middleware await issue and add tests Co-authored-by: MattB-msft <10568244+MattB-msft@users.noreply.github.com> --- .../fastapi/jwt_authorization_middleware.py | 2 +- tests/hosting_fastapi/__init__.py | 2 + .../test_jwt_authorization_middleware.py | 165 ++++++++++++++++++ 3 files changed, 168 insertions(+), 1 deletion(-) create mode 100644 tests/hosting_fastapi/__init__.py create mode 100644 tests/hosting_fastapi/test_jwt_authorization_middleware.py diff --git a/libraries/microsoft-agents-hosting-fastapi/microsoft_agents/hosting/fastapi/jwt_authorization_middleware.py b/libraries/microsoft-agents-hosting-fastapi/microsoft_agents/hosting/fastapi/jwt_authorization_middleware.py index 2a186476..992b91ae 100644 --- a/libraries/microsoft-agents-hosting-fastapi/microsoft_agents/hosting/fastapi/jwt_authorization_middleware.py +++ b/libraries/microsoft-agents-hosting-fastapi/microsoft_agents/hosting/fastapi/jwt_authorization_middleware.py @@ -63,7 +63,7 @@ async def __call__(self, scope: Scope, receive: Receive, send: Send): else: if not auth_config or not auth_config.CLIENT_ID: request.state.claims_identity = ( - await token_validator.get_anonymous_claims() + token_validator.get_anonymous_claims() ) else: response = JSONResponse( diff --git a/tests/hosting_fastapi/__init__.py b/tests/hosting_fastapi/__init__.py new file mode 100644 index 00000000..5b7f7a92 --- /dev/null +++ b/tests/hosting_fastapi/__init__.py @@ -0,0 +1,2 @@ +# Copyright (c) Microsoft Corporation. All rights reserved. +# Licensed under the MIT License. diff --git a/tests/hosting_fastapi/test_jwt_authorization_middleware.py b/tests/hosting_fastapi/test_jwt_authorization_middleware.py new file mode 100644 index 00000000..a2d42bb5 --- /dev/null +++ b/tests/hosting_fastapi/test_jwt_authorization_middleware.py @@ -0,0 +1,165 @@ +# Copyright (c) Microsoft Corporation. All rights reserved. +# Licensed under the MIT License. + +import pytest +from fastapi import FastAPI +from fastapi.testclient import TestClient +from unittest.mock import Mock, AsyncMock, patch + +from microsoft_agents.hosting.core import AgentAuthConfiguration +from microsoft_agents.hosting.core.authorization import ClaimsIdentity, JwtTokenValidator +from microsoft_agents.hosting.fastapi import JwtAuthorizationMiddleware + + +@pytest.fixture +def app_with_middleware(): + """Create a FastAPI app with JWT middleware for testing""" + from fastapi import Request + + app = FastAPI() + app.add_middleware(JwtAuthorizationMiddleware) + + @app.get("/test") + async def test_endpoint(request: Request): + claims_identity = getattr(request.state, "claims_identity", None) + return { + "authenticated": claims_identity.is_authenticated if claims_identity else False, + "claims": claims_identity.claims if claims_identity else {} + } + + return app + + +@pytest.fixture +def auth_config_anonymous(): + """Create auth config that allows anonymous access""" + return AgentAuthConfiguration( + client_id=None, # No client_id means anonymous is allowed + tenant_id=None + ) + + +@pytest.fixture +def auth_config_with_client_id(): + """Create auth config with client_id (requires authentication)""" + return AgentAuthConfiguration( + client_id="test-client-id", + tenant_id="test-tenant-id" + ) + + +class TestJwtAuthorizationMiddleware: + """Tests for JWT Authorization Middleware""" + + def test_anonymous_claims_without_await(self, app_with_middleware, auth_config_anonymous): + """ + Test that get_anonymous_claims() is called without await when no auth header is present + and anonymous access is allowed (no CLIENT_ID configured). + + This is the primary test for the bug fix - ensuring that get_anonymous_claims() + is not awaited since it's a synchronous method. + """ + # Set up the app state with anonymous auth config + app_with_middleware.state.agent_configuration = auth_config_anonymous + + # Mock the JwtTokenValidator.get_anonymous_claims to verify it's called correctly + with patch.object(JwtTokenValidator, 'get_anonymous_claims') as mock_get_anonymous: + mock_claims = ClaimsIdentity({}, False, authentication_type="Anonymous") + mock_get_anonymous.return_value = mock_claims + + # Make request without Authorization header + client = TestClient(app_with_middleware) + response = client.get("/test") + + # Verify the response + assert response.status_code == 200 + assert response.json()["authenticated"] == False + assert response.json()["claims"] == {} + + # Verify get_anonymous_claims was called + mock_get_anonymous.assert_called_once() + + def test_missing_auth_header_with_client_id_returns_401(self, app_with_middleware, auth_config_with_client_id): + """ + Test that missing Authorization header returns 401 when CLIENT_ID is configured + """ + app_with_middleware.state.agent_configuration = auth_config_with_client_id + + client = TestClient(app_with_middleware) + response = client.get("/test") + + assert response.status_code == 401 + assert "Authorization header not found" in response.json()["error"] + + def test_invalid_auth_header_format_returns_401(self, app_with_middleware, auth_config_with_client_id): + """ + Test that invalid Authorization header format returns 401 + """ + app_with_middleware.state.agent_configuration = auth_config_with_client_id + + client = TestClient(app_with_middleware) + response = client.get("/test", headers={"Authorization": "InvalidFormat"}) + + assert response.status_code == 401 + assert "Invalid authorization header format" in response.json()["error"] + + def test_valid_token_sets_claims_identity(self, app_with_middleware, auth_config_with_client_id): + """ + Test that a valid JWT token is validated and claims are set + """ + app_with_middleware.state.agent_configuration = auth_config_with_client_id + + # Mock the validate_token method + with patch.object(JwtTokenValidator, 'validate_token', new_callable=AsyncMock) as mock_validate: + mock_claims = ClaimsIdentity( + {"aud": "test-client-id", "sub": "test-user"}, + True, + security_token="test-token" + ) + mock_validate.return_value = mock_claims + + client = TestClient(app_with_middleware) + response = client.get("/test", headers={"Authorization": "Bearer test-token"}) + + assert response.status_code == 200 + assert response.json()["authenticated"] == True + assert response.json()["claims"]["aud"] == "test-client-id" + + # Verify validate_token was called with the correct token + mock_validate.assert_called_once_with("test-token") + + def test_invalid_token_returns_401(self, app_with_middleware, auth_config_with_client_id): + """ + Test that an invalid JWT token returns 401 + """ + app_with_middleware.state.agent_configuration = auth_config_with_client_id + + # Mock the validate_token method to raise ValueError + with patch.object(JwtTokenValidator, 'validate_token', new_callable=AsyncMock) as mock_validate: + mock_validate.side_effect = ValueError("Invalid token") + + client = TestClient(app_with_middleware) + response = client.get("/test", headers={"Authorization": "Bearer invalid-token"}) + + assert response.status_code == 401 + assert "Invalid token or authentication failed" in response.json()["error"] + + def test_anonymous_access_with_no_auth_config(self, app_with_middleware): + """ + Test that anonymous access is allowed when auth_config is None + """ + # Don't set agent_configuration on the app state (simulating no config) + # This will result in auth_config being None + + with patch.object(JwtTokenValidator, 'get_anonymous_claims') as mock_get_anonymous: + mock_claims = ClaimsIdentity({}, False, authentication_type="Anonymous") + mock_get_anonymous.return_value = mock_claims + + client = TestClient(app_with_middleware) + response = client.get("/test") + + assert response.status_code == 200 + assert response.json()["authenticated"] == False + + # Verify get_anonymous_claims was called + mock_get_anonymous.assert_called_once() From 3439340b4abf83a2d1fb4a3c7c37bc9e179fe81b Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Sun, 1 Feb 2026 01:19:52 +0000 Subject: [PATCH 3/4] Apply code formatting with black and fix flake8 issues Co-authored-by: MattB-msft <10568244+MattB-msft@users.noreply.github.com> --- .../fastapi/jwt_authorization_middleware.py | 4 +- .../test_jwt_authorization_middleware.py | 117 +++++++++++------- 2 files changed, 72 insertions(+), 49 deletions(-) diff --git a/libraries/microsoft-agents-hosting-fastapi/microsoft_agents/hosting/fastapi/jwt_authorization_middleware.py b/libraries/microsoft-agents-hosting-fastapi/microsoft_agents/hosting/fastapi/jwt_authorization_middleware.py index 992b91ae..7859e537 100644 --- a/libraries/microsoft-agents-hosting-fastapi/microsoft_agents/hosting/fastapi/jwt_authorization_middleware.py +++ b/libraries/microsoft-agents-hosting-fastapi/microsoft_agents/hosting/fastapi/jwt_authorization_middleware.py @@ -62,9 +62,7 @@ async def __call__(self, scope: Scope, receive: Receive, send: Send): return else: if not auth_config or not auth_config.CLIENT_ID: - request.state.claims_identity = ( - token_validator.get_anonymous_claims() - ) + request.state.claims_identity = token_validator.get_anonymous_claims() else: response = JSONResponse( {"error": "Authorization header not found"}, diff --git a/tests/hosting_fastapi/test_jwt_authorization_middleware.py b/tests/hosting_fastapi/test_jwt_authorization_middleware.py index a2d42bb5..09f869f5 100644 --- a/tests/hosting_fastapi/test_jwt_authorization_middleware.py +++ b/tests/hosting_fastapi/test_jwt_authorization_middleware.py @@ -4,10 +4,13 @@ import pytest from fastapi import FastAPI from fastapi.testclient import TestClient -from unittest.mock import Mock, AsyncMock, patch +from unittest.mock import AsyncMock, patch from microsoft_agents.hosting.core import AgentAuthConfiguration -from microsoft_agents.hosting.core.authorization import ClaimsIdentity, JwtTokenValidator +from microsoft_agents.hosting.core.authorization import ( + ClaimsIdentity, + JwtTokenValidator, +) from microsoft_agents.hosting.fastapi import JwtAuthorizationMiddleware @@ -15,18 +18,20 @@ def app_with_middleware(): """Create a FastAPI app with JWT middleware for testing""" from fastapi import Request - + app = FastAPI() app.add_middleware(JwtAuthorizationMiddleware) - + @app.get("/test") async def test_endpoint(request: Request): claims_identity = getattr(request.state, "claims_identity", None) return { - "authenticated": claims_identity.is_authenticated if claims_identity else False, - "claims": claims_identity.claims if claims_identity else {} + "authenticated": ( + claims_identity.is_authenticated if claims_identity else False + ), + "claims": claims_identity.claims if claims_identity else {}, } - + return app @@ -34,8 +39,7 @@ async def test_endpoint(request: Request): def auth_config_anonymous(): """Create auth config that allows anonymous access""" return AgentAuthConfiguration( - client_id=None, # No client_id means anonymous is allowed - tenant_id=None + client_id=None, tenant_id=None # No client_id means anonymous is allowed ) @@ -43,104 +47,123 @@ def auth_config_anonymous(): def auth_config_with_client_id(): """Create auth config with client_id (requires authentication)""" return AgentAuthConfiguration( - client_id="test-client-id", - tenant_id="test-tenant-id" + client_id="test-client-id", tenant_id="test-tenant-id" ) class TestJwtAuthorizationMiddleware: """Tests for JWT Authorization Middleware""" - def test_anonymous_claims_without_await(self, app_with_middleware, auth_config_anonymous): + def test_anonymous_claims_without_await( + self, app_with_middleware, auth_config_anonymous + ): """ Test that get_anonymous_claims() is called without await when no auth header is present and anonymous access is allowed (no CLIENT_ID configured). - + This is the primary test for the bug fix - ensuring that get_anonymous_claims() is not awaited since it's a synchronous method. """ # Set up the app state with anonymous auth config app_with_middleware.state.agent_configuration = auth_config_anonymous - + # Mock the JwtTokenValidator.get_anonymous_claims to verify it's called correctly - with patch.object(JwtTokenValidator, 'get_anonymous_claims') as mock_get_anonymous: + with patch.object( + JwtTokenValidator, "get_anonymous_claims" + ) as mock_get_anonymous: mock_claims = ClaimsIdentity({}, False, authentication_type="Anonymous") mock_get_anonymous.return_value = mock_claims - + # Make request without Authorization header client = TestClient(app_with_middleware) response = client.get("/test") - + # Verify the response assert response.status_code == 200 - assert response.json()["authenticated"] == False + assert response.json()["authenticated"] is False assert response.json()["claims"] == {} - + # Verify get_anonymous_claims was called mock_get_anonymous.assert_called_once() - def test_missing_auth_header_with_client_id_returns_401(self, app_with_middleware, auth_config_with_client_id): + def test_missing_auth_header_with_client_id_returns_401( + self, app_with_middleware, auth_config_with_client_id + ): """ Test that missing Authorization header returns 401 when CLIENT_ID is configured """ app_with_middleware.state.agent_configuration = auth_config_with_client_id - + client = TestClient(app_with_middleware) response = client.get("/test") - + assert response.status_code == 401 assert "Authorization header not found" in response.json()["error"] - def test_invalid_auth_header_format_returns_401(self, app_with_middleware, auth_config_with_client_id): + def test_invalid_auth_header_format_returns_401( + self, app_with_middleware, auth_config_with_client_id + ): """ Test that invalid Authorization header format returns 401 """ app_with_middleware.state.agent_configuration = auth_config_with_client_id - + client = TestClient(app_with_middleware) response = client.get("/test", headers={"Authorization": "InvalidFormat"}) - + assert response.status_code == 401 assert "Invalid authorization header format" in response.json()["error"] - def test_valid_token_sets_claims_identity(self, app_with_middleware, auth_config_with_client_id): + def test_valid_token_sets_claims_identity( + self, app_with_middleware, auth_config_with_client_id + ): """ Test that a valid JWT token is validated and claims are set """ app_with_middleware.state.agent_configuration = auth_config_with_client_id - + # Mock the validate_token method - with patch.object(JwtTokenValidator, 'validate_token', new_callable=AsyncMock) as mock_validate: + with patch.object( + JwtTokenValidator, "validate_token", new_callable=AsyncMock + ) as mock_validate: mock_claims = ClaimsIdentity( {"aud": "test-client-id", "sub": "test-user"}, True, - security_token="test-token" + security_token="test-token", ) mock_validate.return_value = mock_claims - + client = TestClient(app_with_middleware) - response = client.get("/test", headers={"Authorization": "Bearer test-token"}) - + response = client.get( + "/test", headers={"Authorization": "Bearer test-token"} + ) + assert response.status_code == 200 - assert response.json()["authenticated"] == True + assert response.json()["authenticated"] is True assert response.json()["claims"]["aud"] == "test-client-id" - + # Verify validate_token was called with the correct token mock_validate.assert_called_once_with("test-token") - def test_invalid_token_returns_401(self, app_with_middleware, auth_config_with_client_id): + def test_invalid_token_returns_401( + self, app_with_middleware, auth_config_with_client_id + ): """ Test that an invalid JWT token returns 401 """ app_with_middleware.state.agent_configuration = auth_config_with_client_id - + # Mock the validate_token method to raise ValueError - with patch.object(JwtTokenValidator, 'validate_token', new_callable=AsyncMock) as mock_validate: + with patch.object( + JwtTokenValidator, "validate_token", new_callable=AsyncMock + ) as mock_validate: mock_validate.side_effect = ValueError("Invalid token") - + client = TestClient(app_with_middleware) - response = client.get("/test", headers={"Authorization": "Bearer invalid-token"}) - + response = client.get( + "/test", headers={"Authorization": "Bearer invalid-token"} + ) + assert response.status_code == 401 assert "Invalid token or authentication failed" in response.json()["error"] @@ -150,16 +173,18 @@ def test_anonymous_access_with_no_auth_config(self, app_with_middleware): """ # Don't set agent_configuration on the app state (simulating no config) # This will result in auth_config being None - - with patch.object(JwtTokenValidator, 'get_anonymous_claims') as mock_get_anonymous: + + with patch.object( + JwtTokenValidator, "get_anonymous_claims" + ) as mock_get_anonymous: mock_claims = ClaimsIdentity({}, False, authentication_type="Anonymous") mock_get_anonymous.return_value = mock_claims - + client = TestClient(app_with_middleware) response = client.get("/test") - + assert response.status_code == 200 - assert response.json()["authenticated"] == False - + assert response.json()["authenticated"] is False + # Verify get_anonymous_claims was called mock_get_anonymous.assert_called_once() From 7372c87e2a4fb23ab6f435fa7bb8865bd972609e Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Sun, 1 Feb 2026 01:28:45 +0000 Subject: [PATCH 4/4] Remove test files as requested Co-authored-by: MattB-msft <10568244+MattB-msft@users.noreply.github.com> --- tests/hosting_fastapi/__init__.py | 2 - .../test_jwt_authorization_middleware.py | 190 ------------------ 2 files changed, 192 deletions(-) delete mode 100644 tests/hosting_fastapi/__init__.py delete mode 100644 tests/hosting_fastapi/test_jwt_authorization_middleware.py diff --git a/tests/hosting_fastapi/__init__.py b/tests/hosting_fastapi/__init__.py deleted file mode 100644 index 5b7f7a92..00000000 --- a/tests/hosting_fastapi/__init__.py +++ /dev/null @@ -1,2 +0,0 @@ -# Copyright (c) Microsoft Corporation. All rights reserved. -# Licensed under the MIT License. diff --git a/tests/hosting_fastapi/test_jwt_authorization_middleware.py b/tests/hosting_fastapi/test_jwt_authorization_middleware.py deleted file mode 100644 index 09f869f5..00000000 --- a/tests/hosting_fastapi/test_jwt_authorization_middleware.py +++ /dev/null @@ -1,190 +0,0 @@ -# Copyright (c) Microsoft Corporation. All rights reserved. -# Licensed under the MIT License. - -import pytest -from fastapi import FastAPI -from fastapi.testclient import TestClient -from unittest.mock import AsyncMock, patch - -from microsoft_agents.hosting.core import AgentAuthConfiguration -from microsoft_agents.hosting.core.authorization import ( - ClaimsIdentity, - JwtTokenValidator, -) -from microsoft_agents.hosting.fastapi import JwtAuthorizationMiddleware - - -@pytest.fixture -def app_with_middleware(): - """Create a FastAPI app with JWT middleware for testing""" - from fastapi import Request - - app = FastAPI() - app.add_middleware(JwtAuthorizationMiddleware) - - @app.get("/test") - async def test_endpoint(request: Request): - claims_identity = getattr(request.state, "claims_identity", None) - return { - "authenticated": ( - claims_identity.is_authenticated if claims_identity else False - ), - "claims": claims_identity.claims if claims_identity else {}, - } - - return app - - -@pytest.fixture -def auth_config_anonymous(): - """Create auth config that allows anonymous access""" - return AgentAuthConfiguration( - client_id=None, tenant_id=None # No client_id means anonymous is allowed - ) - - -@pytest.fixture -def auth_config_with_client_id(): - """Create auth config with client_id (requires authentication)""" - return AgentAuthConfiguration( - client_id="test-client-id", tenant_id="test-tenant-id" - ) - - -class TestJwtAuthorizationMiddleware: - """Tests for JWT Authorization Middleware""" - - def test_anonymous_claims_without_await( - self, app_with_middleware, auth_config_anonymous - ): - """ - Test that get_anonymous_claims() is called without await when no auth header is present - and anonymous access is allowed (no CLIENT_ID configured). - - This is the primary test for the bug fix - ensuring that get_anonymous_claims() - is not awaited since it's a synchronous method. - """ - # Set up the app state with anonymous auth config - app_with_middleware.state.agent_configuration = auth_config_anonymous - - # Mock the JwtTokenValidator.get_anonymous_claims to verify it's called correctly - with patch.object( - JwtTokenValidator, "get_anonymous_claims" - ) as mock_get_anonymous: - mock_claims = ClaimsIdentity({}, False, authentication_type="Anonymous") - mock_get_anonymous.return_value = mock_claims - - # Make request without Authorization header - client = TestClient(app_with_middleware) - response = client.get("/test") - - # Verify the response - assert response.status_code == 200 - assert response.json()["authenticated"] is False - assert response.json()["claims"] == {} - - # Verify get_anonymous_claims was called - mock_get_anonymous.assert_called_once() - - def test_missing_auth_header_with_client_id_returns_401( - self, app_with_middleware, auth_config_with_client_id - ): - """ - Test that missing Authorization header returns 401 when CLIENT_ID is configured - """ - app_with_middleware.state.agent_configuration = auth_config_with_client_id - - client = TestClient(app_with_middleware) - response = client.get("/test") - - assert response.status_code == 401 - assert "Authorization header not found" in response.json()["error"] - - def test_invalid_auth_header_format_returns_401( - self, app_with_middleware, auth_config_with_client_id - ): - """ - Test that invalid Authorization header format returns 401 - """ - app_with_middleware.state.agent_configuration = auth_config_with_client_id - - client = TestClient(app_with_middleware) - response = client.get("/test", headers={"Authorization": "InvalidFormat"}) - - assert response.status_code == 401 - assert "Invalid authorization header format" in response.json()["error"] - - def test_valid_token_sets_claims_identity( - self, app_with_middleware, auth_config_with_client_id - ): - """ - Test that a valid JWT token is validated and claims are set - """ - app_with_middleware.state.agent_configuration = auth_config_with_client_id - - # Mock the validate_token method - with patch.object( - JwtTokenValidator, "validate_token", new_callable=AsyncMock - ) as mock_validate: - mock_claims = ClaimsIdentity( - {"aud": "test-client-id", "sub": "test-user"}, - True, - security_token="test-token", - ) - mock_validate.return_value = mock_claims - - client = TestClient(app_with_middleware) - response = client.get( - "/test", headers={"Authorization": "Bearer test-token"} - ) - - assert response.status_code == 200 - assert response.json()["authenticated"] is True - assert response.json()["claims"]["aud"] == "test-client-id" - - # Verify validate_token was called with the correct token - mock_validate.assert_called_once_with("test-token") - - def test_invalid_token_returns_401( - self, app_with_middleware, auth_config_with_client_id - ): - """ - Test that an invalid JWT token returns 401 - """ - app_with_middleware.state.agent_configuration = auth_config_with_client_id - - # Mock the validate_token method to raise ValueError - with patch.object( - JwtTokenValidator, "validate_token", new_callable=AsyncMock - ) as mock_validate: - mock_validate.side_effect = ValueError("Invalid token") - - client = TestClient(app_with_middleware) - response = client.get( - "/test", headers={"Authorization": "Bearer invalid-token"} - ) - - assert response.status_code == 401 - assert "Invalid token or authentication failed" in response.json()["error"] - - def test_anonymous_access_with_no_auth_config(self, app_with_middleware): - """ - Test that anonymous access is allowed when auth_config is None - """ - # Don't set agent_configuration on the app state (simulating no config) - # This will result in auth_config being None - - with patch.object( - JwtTokenValidator, "get_anonymous_claims" - ) as mock_get_anonymous: - mock_claims = ClaimsIdentity({}, False, authentication_type="Anonymous") - mock_get_anonymous.return_value = mock_claims - - client = TestClient(app_with_middleware) - response = client.get("/test") - - assert response.status_code == 200 - assert response.json()["authenticated"] is False - - # Verify get_anonymous_claims was called - mock_get_anonymous.assert_called_once()