test: dpop

pyeudiw/jwt/__init__.py

@@ -12,6 +12,7 @@
 
 from pyeudiw.jwk import JWK
 from pyeudiw.jwt.utils import unpad_jwt_header
+from pyeudiw.oauth2.exceptions import KidError
 
 DEFAULT_HASH_FUNC = "SHA-256"
 
@@ -115,8 +116,8 @@ def verify(self, jws: str, **kwargs):
 
         _head = unpad_jwt_header(jws)
         if _head.get("kid") != self.jwk.as_dict()["kid"]:  # pragma: no cover
-            raise Exception(
-                f"kid error: {_head.get('kid')} != {self.jwk.as_dict()['kid']}"
+            raise KidError(
+                f"{_head.get('kid')} != {self.jwk.as_dict()['kid']}"
             )
 
         _head["alg"]

pyeudiw/oauth2/dpop.py

@@ -5,6 +5,7 @@
 from pydantic import BaseModel, HttpUrl
 from pyeudiw.jwt import JWSHelper
 from pyeudiw.jwt.utils import unpad_jwt_payload, unpad_jwt_header
+from pyeudiw.oauth2.exceptions import KidError
 from pyeudiw.tools.utils import iat_now
 from typing import Literal
 
@@ -63,7 +64,6 @@ def proof(self):
             }
         )
         return jwt
-        # TODO assertion
 
 
 class DPoPVerifier:
@@ -81,12 +81,36 @@ def __init__(
             if self.dpop_header_prefix in http_header_authz
             else http_header_authz
         )
+        # If the jwt is invalid, this will raise an exception
+        try:
+            unpad_jwt_header(http_header_dpop)
+        except UnicodeDecodeError:
+            raise ValueError("DPoP proof is not a valid JWT")
+        except Exception as e:
+            logger.error(
+                "DPoP proof validation error, "
+                f"{e.__class__.__name__}: {e}"
+            )
+            raise ValueError("DPoP proof is not a valid JWT")
         self.proof = http_header_dpop
 
     @property
     def is_valid(self):
         jws_verifier = JWSHelper(self.public_jwk)
-        dpop_valid = jws_verifier.verify(self.proof)
+        try:
+            dpop_valid = jws_verifier.verify(self.proof)
+        except KidError as e:
+            logger.error(
+                "DPoP proof validation error, "
+                f"kid does not match: {e}"
+            )
+            return False
+        except Exception as e:
+            logger.error(
+                "DPoP proof validation error, "
+                f"{e.__class__.__name__}: {e}"
+            )
+            return False
 
         header = unpad_jwt_header(self.proof)
         DPoPTokenHeaderSchema(**header)

pyeudiw/oauth2/exceptions.py

@@ -0,0 +1,2 @@
+class KidError(Exception):
+    pass

pyeudiw/tests/oauth2/test_dpop.py

@@ -1,6 +1,8 @@
+import hashlib
+
 import pytest
 
-from pyeudiw.oauth2.dpop import DPoPIssuer, DPoPVerifier
+from pyeudiw.oauth2.dpop import DPoPIssuer, DPoPVerifier, DPoPTokenPayloadSchema
 from pyeudiw.jwk import JWK
 from pyeudiw.jwt import JWSHelper
 from pyeudiw.jwt.utils import unpad_jwt_payload, unpad_jwt_header
@@ -65,25 +67,63 @@ def wia_jws(jwshelper):
 
 def test_create_validate_dpop_http_headers(wia_jws, private_jwk=PRIVATE_JWK):
     # create
-    unpad_jwt_header(wia_jws)
-    payload = unpad_jwt_payload(wia_jws)
-    # TODO assertions for upadded headers and payload
+    header = unpad_jwt_header(wia_jws)
+    assert header
+    assert isinstance(header["trust_chain"], list)
+    assert isinstance(header["x5c"], list)
+    assert header["alg"]
+    assert header["kid"]
 
     new_dpop = DPoPIssuer(
         htu='https://example.org/redirect',
         token=wia_jws,
         private_jwk=private_jwk
     )
     proof = new_dpop.proof
+    assert proof
+
+    header = unpad_jwt_header(proof)
+    assert header["typ"] == "dpop+jwt"
+    assert header["alg"]
+    assert "mac" not in str(header["alg"]).lower()
+    assert "d" not in header["jwk"]
 
-    # TODO assertions
+    payload = unpad_jwt_payload(proof)
+    assert payload["ath"] == hashlib.sha256(wia_jws.encode()).hexdigest()
+    assert payload["htm"] in ["GET", "POST", "get", "post"]
+    assert payload["htu"] == "https://example.org/redirect"
+    assert payload["jti"]
+    assert payload["iat"]
 
     # verify
     dpop = DPoPVerifier(
-        public_jwk=payload['cnf']['jwk'],
+        public_jwk=PUBLIC_JWK,
         http_header_authz=f"DPoP {wia_jws}",
         http_header_dpop=proof
     )
-
     assert dpop.is_valid
-    # TODO assertions
+
+    other_jwk = JWK(key_type="RSA").public_key
+    dpop = DPoPVerifier(
+        public_jwk=other_jwk,
+        http_header_authz=f"DPoP {wia_jws}",
+        http_header_dpop=proof
+    )
+    assert dpop.is_valid is False
+
+    with pytest.raises(ValueError):
+        dpop = DPoPVerifier(
+            public_jwk=PUBLIC_JWK,
+            http_header_authz=f"DPoP {wia_jws}",
+            http_header_dpop="aaa"
+        )
+        assert dpop.is_valid is False
+
+    with pytest.raises(ValueError):
+        dpop = DPoPVerifier(
+            public_jwk=PUBLIC_JWK,
+            http_header_authz=f"DPoP {wia_jws}",
+            http_header_dpop="aaa" + proof[3:]
+        )
+        assert dpop.is_valid is False
+

pyeudiw/tests/tools/test_jwt.py

@@ -78,7 +78,6 @@ def test_jws_helper_sign(jwk, payload):
     helper = JWSHelper(jwk)
     jws = helper.sign(payload)
     assert jws
-    print(jws)
 
 
 @pytest.mark.parametrize("jwk, payload", JWKs_RSA)