fix: refactor bitstring status list verifier

Signed-off-by: Athan Massouras <[email protected]>

Author: Athan Massouras

src/bitstring_status_list/verifier.py

@@ -4,9 +4,10 @@
     Protocol,
 )
 
-import requests as r
+from aiohttp import ClientSession
+import json
 
-from bit_array import *
+from bit_array import BitArray, b64url_decode, b64url_encode, dict_to_b64
 from bitstring_status_list.issuer import MIN_LIST_LENGTH, StatusListLengthError
 
 class EnvelopingTokenVerifier(Protocol):
@@ -34,47 +35,68 @@ def __init__(
         self,
         credential_status: dict,
 
-        headers: Optional[dict] = None,
-        payload: Optional[dict] = None,
+        headers: Optional[dict],
+        payload: dict,
 
-        bit_array: Optional[BitArray] = None,
+        bit_array: BitArray,
     ):
         self.credential_status = credential_status
-        assert all(key in credential_status.keys() for key in ["id", "type", "statusPurpose", "statusListIndex", "statusListCredential"]),\
-            "Invalid credentialStatus"
+        if not all(key in credential_status.keys() for key in ["id", "type", "statusPurpose", "statusListIndex", "statusListCredential"]):
+            raise StatusVerificationError(f"Invalid credential_status: {credential_status}. \
+                                            credential status is expected to have keys: \
+                                            [id, type, statusPurpose, statusListIndex, statusListCredential]")
 
         self.headers = headers
         self.payload = payload
 
         self._bit_array = bit_array
 
-    def establish_connection(
-        self, 
-        status_list_format: Literal["CWT", "JWT"],
-    ) -> bytes:
-        """ Establish connection. Returns base64 encoded response. """
-        issuer_uri = self.credential_status["statusListCredential"]
-        try:
-            response = r.get(issuer_uri)
-        except Exception as e:
-            raise StatusRetrievalError(f"Dereference of uri {issuer_uri} failed: {e}.")
-        
-        if not (200 <= response.status_code < 300):
-            raise StatusRetrievalError(f"Response status from {issuer_uri} was {response.status_code}.")
-
-        # When establishing a new connection, clear previous cached values.
-        self.headers = None
-        self.payload = None
-        self._bit_array = None
-
-        return response.content
-
-    def verify_jwt(
-            self, 
-            sl_response: bytes, 
-            verifier: EnvelopingTokenVerifier | EmbeddingTokenVerifier,
-            min_list_length: int = MIN_LIST_LENGTH,
-        ):
+    @classmethod
+    async def retrieve_list(
+        cls, 
+        credential_status: dict,
+        verifier: EnvelopingTokenVerifier | EmbeddingTokenVerifier,
+        headers: dict | None = None,
+        min_list_length: int = MIN_LIST_LENGTH,
+    ) -> "BitstringStatusListVerifier":
+        """ 
+        Establish connection, parse and verify response, and create instance of 
+        BitstringStatusListVerifier to access it.
+
+        Args:
+            credential_status: REQUIRED. The credentialStatus field of a verifiable credential as
+            specified in S. 2.1.
+
+            verifier: REQUIRED. A callable that verifies the signature of a payload, equivalent to 
+            signer in sign_jwt() in issuer.py.
+
+            headers: OPTIONAL. Additional headers for the HTTP request.
+
+            min_list_length: OPTIONAL. The minimum list length, recommended to be 131,072 (see S. 6.1)
+        Returns:
+            An instance of BitstringStatusListVerifier which has been verified for correctness and 
+            integrity.
+        """
+
+        headers = headers or {}
+
+        async with ClientSession() as session:
+            async with session.get(credential_status["statusListCredential"], headers=headers) as resp:
+                if not 200 <= resp.status < 300:
+                    raise StatusRetrievalError(f"Unable to retrieve token at {credential_status["statusListCredential"]}")
+                
+                token = await resp.read()
+
+        return cls.from_jwt(token, credential_status, verifier, min_list_length)
+
+    @classmethod
+    def from_jwt(
+        cls, 
+        token: bytes | str, 
+        credential_status: dict,
+        verifier: EnvelopingTokenVerifier | EmbeddingTokenVerifier,
+        min_list_length: int = MIN_LIST_LENGTH,
+    ) -> "BitstringStatusListVerifier":
         """ 
         Takes a status-list response and a verifier, and ensures that the response matches the 
         required format, verifying the signature using verifier.
@@ -83,65 +105,81 @@ def verify_jwt(
         signature is correct, and raise an exception if not.
 
         Args:
-            sl_response: REQUIRED. A base64-encoded status_list response, acquired (eg.) from 
+            token: REQUIRED. A base64-encoded status_list response, acquired (eg.) from 
             establish_connection().
 
+            credential_status: REQUIRED. The credentialStatus field of a verifiable credential as
+            specified in S. 2.1.
+
             verifier: REQUIRED. A callable that verifies the signature of a payload. Must match
-            the proof format of the sl_response (embedded or enveloping)
+            the proof format of the token (embedded or enveloping)
 
             min_list_length: OPTIONAL. The minimum list length, recommended to be 131,072 (see S. 6.1)
         """
-        
-        if b"." in sl_response:
+        # Check that message is in valid JWT format 
+        if isinstance(token, str):
+            token = token.encode()
+
+        if not token.startswith(b"ey"):
+            raise ValueError("JWT requested but token is not a JWT") 
+
+        headers = None
+        if b"." in token:
             # Enveloping proof
 
             # Check that message is in valid JWT format 
-            headers_bytes, payload_bytes, signature = sl_response.split(b".", maxsplit=3)
+            headers_bytes, payload_bytes, signature = token.split(b".", maxsplit=3)
             assert headers_bytes and payload_bytes and signature
 
             # Verify signature. verifier must be of type EnvelopingTokenVerifier
             if not verifier(headers_bytes + b"." + payload_bytes, signature):
                 raise StatusVerificationError("Invalid signature on payload.")
 
             # Extract data
-            self.headers = json.loads(b64url_decode(headers_bytes))
-            self.payload = json.loads(b64url_decode(payload_bytes))
+            headers = json.loads(b64url_decode(headers_bytes))
+            payload = json.loads(b64url_decode(payload_bytes))
         else:
             # Embedding proof
 
             # Extract data
-            self.payload = json.loads(b64url_decode(sl_response))
+            payload = json.loads(b64url_decode(token))
 
             # Verify signature
-            unsigned_payload = {key: self.payload[key] for key in self.payload if key != "proof"}
-            if not verifier(dict_to_b64(unsigned_payload), self.payload["proof"]):
+            unsigned_payload = {key: payload[key] for key in payload if key != "proof"}
+            if not verifier(dict_to_b64(unsigned_payload), payload["proof"]):
                 raise StatusVerificationError("Invalid signature on payload")
 
         # Check values of status list against provided credential
-        credential_subject = self.payload["credentialSubject"]
-        if credential_subject["statusPurpose"] != self.credential_status["statusPurpose"]:
+        credential_subject = payload["credentialSubject"]
+        if credential_subject["statusPurpose"] != credential_status["statusPurpose"]:
             raise StatusVerificationError(
-                f"statusPurpose in credential is {self.credential_status["statusPurpose"]}, while \
+                f"statusPurpose in credential is {credential_status["statusPurpose"]}, while \
                 statusPurpose in status list is {credential_subject["statusPurpose"]}"
             )
 
         # If statusPurpose = message, ensure that a statusMessage list exists in the credential
-        bits = self.credential_status.get("statusSize")
-        if bits is not None and bits > 1 and self.credential_status.get("statusMessage") is None:
+        bits = credential_status.get("statusSize")
+        if bits is not None and bits > 1 and credential_status.get("statusMessage") is None:
             raise StatusVerificationError("For statusSize > 1, a message must exist.")
 
-        if self.credential_status["statusPurpose"] == "message" and self.credential_status.get("statusMessage") is None:
+        if credential_status["statusPurpose"] == "message" and credential_status.get("statusMessage") is None:
             raise StatusVerificationError("If statusPurpose is `message`, a statusMessage field must \
                                           be included which provides the message associated with each bit.")
 
         # Cache returned status list as BitArray
-        self._bit_array = BitArray.from_b64(1 if bits is None else bits, credential_subject["encodedList"])
-        if self._bit_array.size < min_list_length:
+        bit_array = BitArray.from_b64(1 if bits is None else bits, credential_subject["encodedList"])
+        if bit_array.size < min_list_length:
             raise StatusListLengthError(f"Bitstring status list must be at least {min_list_length} \
-                                        bits long, but was {self._bit_array.size} bits long instead.")
+                                        bits long, but was {bit_array.size} bits long instead.")
+        
+        return cls(
+            credential_status=credential_status,
+            headers=headers,
+            payload=payload,
+            bit_array=bit_array,
+        )
 
     def get_status(self, idx: Optional[int] = None):
-        assert self._bit_array is not None, "Before accessing the status, please verify using jwt_verify or cwt_verify"
         if idx is None:
             idx = int(self.credential_status["statusListIndex"])
 

src/token_status_list/verifier.py

@@ -56,8 +56,22 @@ async def retrieve_list(
         headers: dict | None = None,
     ) -> "TokenStatusListVerifier":
         """ 
-        Establish connection. Parse and verify response, and create instance of 
-        TokenStatusListVerifier to access it. 
+        Establish connection, parse and verify response, and create instance of 
+        TokenStatusListVerifier to access it.
+
+        Args:
+            status_list_uri: REQUIRED. The uri where the status list can be accessed via HTTP request.
+
+            verifier: REQUIRED. A callable that verifies the signature of a payload, equivalent to 
+            signer in sign_jwt() in issuer.py.
+
+            encoding: OPTIONAL. Either JWT or CWT. Default is JWT.
+
+            headers: OPTIONAL. Additional headers for the HTTP request that is sent to status_list_uri.
+
+        Returns:
+            An instance of TokenStatusListVerifier which has been verified for correctness and 
+            integrity.
         """
 
         headers = headers or {}

tests/test_bitstring_status_list.py

@@ -77,8 +77,11 @@ def test_verify_jwt_basic_enveloping(status: BitstringStatusListIssuer):
         "statusListCredential": "https://example.com/credentials/status/3"
     }
 
-    verifier = BitstringStatusListVerifier(credential_status)
-    verifier.verify_jwt(encoded_jwt, verifier=trivial_enveloping_verifier)
+    verifier = BitstringStatusListVerifier.from_jwt(
+        token=encoded_jwt,
+        credential_status=credential_status,
+        verifier=trivial_embedding_verifier,
+    )
 
     assert verifier.headers == {"alg": "ES256", "kid": "12"}
     assert verifier.payload == {
@@ -115,8 +118,11 @@ def test_verify_jwt_basic_embedding(status: BitstringStatusListIssuer):
         "statusListCredential": "https://example.com/credentials/status/3"
     }
 
-    verifier = BitstringStatusListVerifier(credential_status)
-    verifier.verify_jwt(encoded_jwt, verifier=trivial_embedding_verifier)
+    verifier = BitstringStatusListVerifier.from_jwt(
+        token=encoded_jwt,
+        credential_status=credential_status,
+        verifier=trivial_embedding_verifier,
+    )
 
     assert verifier.payload == {
         "@context": [
@@ -178,8 +184,11 @@ def test_status_message():
         "statusSize": 2,
     }
 
-    verifier = BitstringStatusListVerifier(credential_status)
-    verifier.verify_jwt(encoded_jwt, verifier=trivial_enveloping_verifier)
+    verifier = BitstringStatusListVerifier.from_jwt(
+        token=encoded_jwt,
+        credential_status=credential_status,
+        verifier=trivial_enveloping_verifier,
+    )
 
     for i in range(bitstring.status_list.size):
         assert verifier.get_status(i) == {
@@ -208,8 +217,11 @@ def test_verify_es256_enveloping(
         "statusListCredential": "https://example.com/credentials/status/3"
     }
 
-    verifier = BitstringStatusListVerifier(credential_status)
-    verifier.verify_jwt(encoded_jwt, verifier=es256_enveloping_verifier)
+    verifier = BitstringStatusListVerifier.from_jwt(
+        token=encoded_jwt,
+        credential_status=credential_status,
+        verifier=es256_enveloping_verifier,
+    ) 
 
     for i in range(status.status_list.size):
         assert verifier.get_status(i) == {
@@ -235,8 +247,11 @@ def test_verify_es256_embedding(
         "statusListCredential": "https://example.com/credentials/status/3"
     }
 
-    verifier = BitstringStatusListVerifier(credential_status)
-    verifier.verify_jwt(encoded_jwt, verifier=es256_embedding_verifier)
+    verifier = BitstringStatusListVerifier.from_jwt(
+        token=encoded_jwt,
+        credential_status=credential_status,
+        verifier=es256_embedding_verifier,
+    )
 
     for i in range(status.status_list.size):
         assert verifier.get_status(i) == {