|
6 | 6 | from unittest.mock import AsyncMock, MagicMock |
7 | 7 |
|
8 | 8 | import pytest |
| 9 | +from jwcrypto import jwk |
9 | 10 |
|
10 | 11 | from auth0_server_python.auth_server.mfa_client import DEFAULT_MFA_TOKEN_TTL, MfaClient |
11 | 12 | from auth0_server_python.auth_types import ( |
@@ -892,3 +893,121 @@ async def test_verify_persist_store_failure_raises(self, mocker): |
892 | 893 | {"mfa_token": _enc(), "otp": "123456", |
893 | 894 | "persist": True, "audience": "https://api.example.com"} |
894 | 895 | ) |
| 896 | + |
| 897 | + @pytest.mark.asyncio |
| 898 | + async def test_verify_dpop_attaches_proof_header(self, mocker): |
| 899 | + """When dpop_key is supplied, a DPoP proof header is sent and a bound token accepted.""" |
| 900 | + client = _make_client() |
| 901 | + dpop_key = jwk.JWK.generate(kty="EC", crv="P-256") |
| 902 | + response = AsyncMock() |
| 903 | + response.status_code = 200 |
| 904 | + response.headers = {} |
| 905 | + response.json = MagicMock(return_value={ |
| 906 | + "access_token": "bound_at", "token_type": "DPoP", "expires_in": 3600 |
| 907 | + }) |
| 908 | + |
| 909 | + captured_request = {} |
| 910 | + |
| 911 | + async def mock_post(self_client, url, **kwargs): |
| 912 | + captured_request["kwargs"] = kwargs |
| 913 | + return response |
| 914 | + |
| 915 | + mocker.patch("httpx.AsyncClient.post", new=mock_post) |
| 916 | + |
| 917 | + result = await client.verify( |
| 918 | + {"mfa_token": _enc(), "otp": "123456"}, |
| 919 | + dpop_key=dpop_key, |
| 920 | + ) |
| 921 | + assert result.token_type == "DPoP" |
| 922 | + assert "DPoP" in captured_request["kwargs"]["headers"] |
| 923 | + |
| 924 | + @pytest.mark.asyncio |
| 925 | + async def test_verify_dpop_nonce_retry(self, mocker): |
| 926 | + """RFC 9449 §8.2: a DPoP-Nonce challenge triggers exactly one retry with the nonce.""" |
| 927 | + client = _make_client() |
| 928 | + dpop_key = jwk.JWK.generate(kty="EC", crv="P-256") |
| 929 | + |
| 930 | + challenge = AsyncMock() |
| 931 | + challenge.status_code = 400 |
| 932 | + challenge.headers = {"DPoP-Nonce": "server-nonce-123"} |
| 933 | + challenge.json = MagicMock(return_value={"error": "use_dpop_nonce"}) |
| 934 | + |
| 935 | + success = AsyncMock() |
| 936 | + success.status_code = 200 |
| 937 | + success.headers = {} |
| 938 | + success.json = MagicMock(return_value={ |
| 939 | + "access_token": "bound_at", "token_type": "DPoP", "expires_in": 3600 |
| 940 | + }) |
| 941 | + |
| 942 | + proofs = [] |
| 943 | + |
| 944 | + async def mock_post(self_client, url, **kwargs): |
| 945 | + proofs.append(kwargs["headers"].get("DPoP")) |
| 946 | + return challenge if len(proofs) == 1 else success |
| 947 | + |
| 948 | + mocker.patch("httpx.AsyncClient.post", new=mock_post) |
| 949 | + |
| 950 | + result = await client.verify( |
| 951 | + {"mfa_token": _enc(), "otp": "123456"}, |
| 952 | + dpop_key=dpop_key, |
| 953 | + ) |
| 954 | + assert result.token_type == "DPoP" |
| 955 | + assert len(proofs) == 2 |
| 956 | + assert proofs[0] != proofs[1] |
| 957 | + |
| 958 | + @pytest.mark.asyncio |
| 959 | + async def test_verify_dpop_rejects_bearer_downgrade(self, mocker): |
| 960 | + """dpop_key supplied but server returns Bearer: reject rather than downgrade.""" |
| 961 | + client = _make_client() |
| 962 | + dpop_key = jwk.JWK.generate(kty="EC", crv="P-256") |
| 963 | + response = AsyncMock() |
| 964 | + response.status_code = 200 |
| 965 | + response.headers = {} |
| 966 | + response.json = MagicMock(return_value={ |
| 967 | + "access_token": "at", "token_type": "Bearer", "expires_in": 3600 |
| 968 | + }) |
| 969 | + mocker.patch("httpx.AsyncClient.post", new_callable=AsyncMock, return_value=response) |
| 970 | + |
| 971 | + with pytest.raises(MfaVerifyError, match="DPoP token binding failed"): |
| 972 | + await client.verify( |
| 973 | + {"mfa_token": _enc(), "otp": "123456"}, |
| 974 | + dpop_key=dpop_key, |
| 975 | + ) |
| 976 | + |
| 977 | + @pytest.mark.asyncio |
| 978 | + async def test_verify_dpop_bound_token_without_key_rejected(self, mocker): |
| 979 | + """Server returns a DPoP-bound token but no key supplied: fail closed, don't accept.""" |
| 980 | + client = _make_client() |
| 981 | + response = AsyncMock() |
| 982 | + response.status_code = 200 |
| 983 | + response.headers = {} |
| 984 | + response.json = MagicMock(return_value={ |
| 985 | + "access_token": "bound_at", "token_type": "DPoP", "expires_in": 3600 |
| 986 | + }) |
| 987 | + mocker.patch("httpx.AsyncClient.post", new_callable=AsyncMock, return_value=response) |
| 988 | + |
| 989 | + with pytest.raises(MfaVerifyError, match="no dpop_key was"): |
| 990 | + await client.verify({"mfa_token": _enc(), "otp": "123456"}) |
| 991 | + |
| 992 | + @pytest.mark.asyncio |
| 993 | + async def test_verify_without_dpop_no_dpop_header(self, mocker): |
| 994 | + """Without dpop_key the request carries no DPoP header and Bearer is accepted.""" |
| 995 | + client = _make_client() |
| 996 | + response = AsyncMock() |
| 997 | + response.status_code = 200 |
| 998 | + response.headers = {} |
| 999 | + response.json = MagicMock(return_value={ |
| 1000 | + "access_token": "at", "token_type": "Bearer", "expires_in": 3600 |
| 1001 | + }) |
| 1002 | + |
| 1003 | + captured_request = {} |
| 1004 | + |
| 1005 | + async def mock_post(self_client, url, **kwargs): |
| 1006 | + captured_request["kwargs"] = kwargs |
| 1007 | + return response |
| 1008 | + |
| 1009 | + mocker.patch("httpx.AsyncClient.post", new=mock_post) |
| 1010 | + |
| 1011 | + result = await client.verify({"mfa_token": _enc(), "otp": "123456"}) |
| 1012 | + assert result.token_type == "Bearer" |
| 1013 | + assert "DPoP" not in captured_request["kwargs"]["headers"] |
0 commit comments