fix: add support for multibit statuses to bitstring status list

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

Author: Athan Massouras

src/bit_array.py

@@ -352,11 +352,8 @@ def rand_and_settle(self):
         byte_idx = self._rand_settle(
             len(self.allocated.lst), lambda index: self.allocated.lst[index] < 255
         )
-        print("byte_idx:\t", byte_idx)
-        print(len(self.allocated.lst))
         start = byte_idx << 3
         end = start + 8
-        print("start, end:\t", start, end)
         index = choice(self.linear_scan(start, end, lambda i: self.allocated[i] == 0))
         self.allocated[index] = 1
         self.num_allocated += 1

src/bitstring_status_list/issuer.py

@@ -102,7 +102,14 @@ def generate_jwt(
         validUntil: Optional[str] = None,
         ttl: Optional[int] = None,
         issuer: Optional[str] = None,
+        status_messages: Optional[list] = None,
+        status_size: Optional[Bits] = None,
     ) -> Tuple[dict, dict]:
+        if status_purpose == "message":
+            assert status_size is not None
+            if status_size > 1:
+                assert status_messages is not None
+        
         headers = {
             "kid": kid,
             "alg": alg,
@@ -130,6 +137,8 @@ def generate_jwt(
                 "statusPurpose": status_purpose,
                 "encodedList": self.status_list.to_b64(),
                 **({"ttl": ttl} if ttl else {}),
+                **({"statusMessages": status_messages} if status_messages else {}),
+                **({"statusSize": status_size} if status_size else {}),
             }
         }
 
@@ -147,6 +156,8 @@ def sign_jwt_enveloping(
         validUntil: Optional[str] = None,
         ttl: Optional[int] = None,
         issuer: Optional[str] = None,
+        status_messages: Optional[list] = None,
+        status_size: Optional[Bits] = None,
     ) -> bytes:
         headers, payload = self.generate_jwt(
             alg=alg,
@@ -158,6 +169,8 @@ def sign_jwt_enveloping(
             validUntil=validUntil,
             ttl=ttl,
             issuer=issuer,
+            status_messages=status_messages,
+            status_size=status_size,
         )
 
         enc_headers = dict_to_b64(headers)
@@ -177,6 +190,8 @@ def sign_jwt_embedding(
         validUntil: Optional[str] = None,
         ttl: Optional[int] = None,
         issuer: Optional[str] = None,
+        status_messages: Optional[list] = None,
+        status_size: Optional[Bits] = None,
     ):
         headers, payload = self.generate_jwt(
             alg=None,
@@ -188,6 +203,8 @@ def sign_jwt_embedding(
             validUntil=validUntil,
             ttl=ttl,
             issuer=issuer,
+            status_messages=status_messages,
+            status_size=status_size,
         )
 
         unsigned_payload_bytes = dict_to_b64(payload)

src/bitstring_status_list/verifier.py

@@ -125,8 +125,16 @@ def verify_jwt(
                 statusPurpose in status list is {credential_subject["statusPurpose"]}"
             )
 
-        # Cache returned status list as BitArray
+        # 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:
+            raise StatusVerificationError("For statusSize > 1, a message must exist.")
+        
+        if self.credential_status["statusPurpose"] == "message" and self.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:
             raise StatusListLengthError(f"Bitstring status list must be at least {min_list_length} \
@@ -147,16 +155,18 @@ def get_status(self, idx: Optional[int] = None):
         # If purpose == message, extract the relevant message and add it to the return_dict, as 
         # described in S. 3.2 Part 14.
         purpose = self.credential_status.get("statusPurpose")
-        if purpose is not None and purpose == "message":
-            try:
-                for message in self.credential_status["statusMessage"]:
-                    if int(message["status"], 16) == status:
-                        return_dict["message"] = message["message"]
-                        break
-
-                raise StatusVerificationError(f"Status not found in message list: {self.credential_status["statusMessage"]}")
-            except KeyError as k: 
-                raise StatusVerificationError(f"statusMessage is malformed or not present: {k}")
-
-        return return_dict
+        if purpose is None or purpose != "message":
+            return return_dict
+        
+        # if purpose == "message"
+        try:
+            for message in self.credential_status["statusMessage"]:
+                if int(message["status"], 16) == status:
+                    return_dict["message"] = message["message"]
+                    return return_dict
+        
+        except KeyError as k: 
+            raise StatusVerificationError(f"statusMessage is malformed or not present: {k}")
+        
+        raise StatusVerificationError(f"Status {status} not found in message list: {self.credential_status["statusMessage"]}")
 

tests/test_bitstring_status_list.py

@@ -122,26 +122,30 @@ def test_status_message():
     for idx in bitstring.take_n(50):
         bitstring[idx] = 3
 
+    status_messages = [
+            {"status":"0x0", "message":"0"},
+            {"status":"0x1", "message":"1"},
+            {"status":"0x2", "message":"2"},
+            {"status":"0x3", "message":"3"},
+    ]
+
     encoded_jwt = bitstring.sign_jwt_enveloping(
         signer=trivial_enveloping_signer,
         alg="ES256",
         kid="12",
-        status_purpose="revocation",
+        status_purpose="message",
+        status_messages=status_messages,
+        status_size=2,
     )
 
-    status_message = [
-        {"status":"0x0", "message":"0"},
-        {"status":"0x1", "message":"1"},
-        {"status":"0x2", "message":"2"},
-        {"status":"0x3", "message":"3"},
-    ]
-    
     credential_status = {
         "id": "https://example.com/credentials/status/3#94567",
         "type": "BitstringStatusListEntry",
-        "statusPurpose": "revocation",
+        "statusPurpose": "message",
         "statusListIndex": "0",
-        "statusListCredential": "https://example.com/credentials/status/3"
+        "statusListCredential": "https://example.com/credentials/status/3",
+        "statusMessage": status_messages,
+        "statusSize": 2,
     }
 
     verifier = BitstringStatusListVerifier(credential_status)

tests/test_token_status_list_verifier.py

@@ -80,7 +80,7 @@ def test_verify_jwt_basic(status: TokenStatusListIssuer):
     }
 
     # Check that statuses match
-    for i in range(len(status)):
+    for i in range(status.status_list.size):
         assert status[i] == verifier.get_status(i)
 
 def test_verify_jwt_expired(status: TokenStatusListIssuer):
@@ -117,7 +117,7 @@ def test_verify_jwt_es256(status: TokenStatusListIssuer, es256_signer, es256_ver
     verifier.jwt_verify(payload.encode(), es256_verifier)
 
     # Check that values match
-    for i in range(len(status)):
+    for i in range(status.status_list.size):
         assert status[i] == verifier.get_status(i)
 
 def test_verify_cwt_basic(status: TokenStatusListIssuer):
@@ -154,7 +154,7 @@ def test_verify_cwt_basic(status: TokenStatusListIssuer):
     }
 
     # Check that values match
-    for i in range(len(status)):
+    for i in range(status.status_list.size):
         assert status[i] == verifier.get_status(i)
 
 def test_verify_cwt_expired(status: TokenStatusListIssuer):
@@ -191,5 +191,5 @@ def test_verify_cwt_es256(status: TokenStatusListIssuer, es256_signer, es256_ver
     verifier.cwt_verify(token, es256_verifier)
 
     # Check that values match
-    for i in range(len(status)):
+    for i in range(status.status_list.size):
         assert status[i] == verifier.get_status(i)