feat: initial verifier for bitstring status list

untested, only supports jwts

Signed-off-by: Athan Massouras <athan@indicio.tech>

src/bitstring_status_list/issuer.py

@@ -17,20 +17,26 @@
 
 MIN_LIST_LENGTH = 131072
 
+class StatusListLengthError(Exception):
+    """Raised when the status list is insufficiently long."""
+
 class BitstringStatusListIssuer(Issuer):
     """Bitstring Status List Issuer."""
     def __init__(
         self,
         status_list: BitArray[N],
         allocator: IndexAllocator,
+        min_list_length: int = MIN_LIST_LENGTH,
     ):
-        super().__init__
-        if self.status_list.bits != 1:
-            raise ValueError("Bitstring status list must have single bit statuses.")
-
-        if len(self.status_list) < MIN_LIST_LENGTH:
-            raise ValueError(f"Bitstring status list must be at least {MIN_LIST_LENGTH} bits long, 
-                             but was {len(self.status_list)} bits long instead.")
+        super().__init__(
+            status_list=status_list,
+            allocator=allocator
+        )
+        self.min_list_length = min_list_length
+
+        if len(self.status_list) < self.min_list_length:
+            raise StatusListLengthError(f"Bitstring status list must be at least {self.min_list_length} 
+                                        bits long, but was {len(self.status_list)} bits long instead.")
 
     @classmethod
     def load(cls, value: dict) -> "BitstringStatusListIssuer":
@@ -57,33 +63,25 @@ def load(cls, value: dict) -> "BitstringStatusListIssuer":
             raise TypeError("status_list must be dict")
 
         parsed_status_list = BitArray.load(status_list)
-
-        if parsed_status_list.bits != 1:
-            raise ValueError("Bitstring status list must have single bit statuses.")
-
-        if len(parsed_status_list) < MIN_LIST_LENGTH:
-            raise ValueError(f"Bitstring status list must be at least {MIN_LIST_LENGTH} bits long, 
-                             but was {len(parsed_status_list)} bits long instead.")
-
         return cls(parsed_status_list, allocator)
 
     @classmethod
-    def new(cls, size: int, strategy: Literal["linear", "random"] = "random") -> "BitstringStatusListIssuer":
+    def new(cls, size: int, bits: Bits = 1, strategy: Literal["linear", "random"] = "random", min_list_length: int = MIN_LIST_LENGTH) -> "BitstringStatusListIssuer":
         """Return a new Issuer."""
-        if size < MIN_LIST_LENGTH:
-            raise ValueError(f"Bitstring status list must be at least {MIN_LIST_LENGTH} bits long, 
-                             but was {size} bits long instead.")
+        if size < min_list_length:
+            raise StatusListLengthError(f"Bitstring status list must be at least {min_list_length} bits 
+                                        long, but was {size} bits long instead.")
 
         if strategy == "linear":
             allocator = LinearIndexAllocator(size)
         elif strategy == "random":
             allocator = RandomIndexAllocator(
-                BitArray.with_at_least(1, size), num_allocated=0
+                BitArray.with_at_least(bits, size), num_allocated=0
             )
         else:
             raise ValueError(f"Invalid strategy: {strategy}")
 
-        status_list = BitArray.with_at_least(1, size)
+        status_list = BitArray.with_at_least(bits, size)
         return cls(status_list, allocator)
 
     def generate_jwt(
@@ -153,9 +151,9 @@ def sign_jwt_enveloping(
             ttl=ttl,
         )
 
-        enc_headers = dict_to_b64(headers).decode()
-        enc_payload = dict_to_b64(payload).decode()
-        enc_to_sign = f"{enc_headers}.{enc_payload}".encode()
+        enc_headers = dict_to_b64(headers)
+        enc_payload = dict_to_b64(payload)
+        enc_to_sign = enc_headers + b"." + enc_payload
 
         signature = signer(enc_to_sign)
         return enc_to_sign + b"." + signature
@@ -183,7 +181,8 @@ def sign_jwt_embedding(
             ttl=ttl,
         )
 
+        payload.update(headers)
         payload["proof"] = proof  # TODO: is this correct?
-        return dict_to_b64(headers) + b"." + dict_to_b64(payload)
+        return dict_to_b64(payload)
 
 

src/bitstring_status_list/verifier.py

@@ -12,6 +12,11 @@
     cast,
 )
 
+import requests as r
+
+from bit_array import *
+from bitstring_status_list.issuer import MIN_LIST_LENGTH, StatusListLengthError
+
 class EnvelopingTokenVerifier(Protocol):
     """Protocol defining the verifying callable for enveloping signatures."""
 
@@ -22,9 +27,139 @@ def __call__(self, payload: bytes, signature: bytes) -> bool:
 class EmbeddingTokenVerifier(Protocol):
     """Protocol defining the verifying callable for embedding signatures."""
 
-    def __call_(self, payload: bytes, signature: dict) -> bool:
+    def __call__(self, payload: bytes, signature: dict) -> bool:
         ...
 
+class StatusRetrievalError(Exception):
+    """Raised if dereference of URL fails. See Bitstring Status List Spec S. 3.2"""
+
+class StatusVerificationError(Exception):
+    """Raised if proofs fail or if format is invalid. See Bitstring Status List Spec S. 3.2"""
+
 class BitstringStatusListVerifier():
-    def __init__(self):
-        pass
+    def __init__(
+        self,
+        credential_status: dict,
+
+        headers: Optional[dict] = None,
+        payload: Optional[dict] = None,
+
+        bit_array: Optional[BitArray] = None,
+    ):
+        self.credential_status = credential_status
+        assert all(key in credential_status.keys() for key in ["id", "type", "statusPurpose", "statusListIndex", "statusListCredential"]),\
+            "Invalid credentialStatus"
+
+        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,
+        ):
+        """ 
+        Takes a status-list response and a verifier, and ensures that the response matches the 
+        required format, verifying the signature using verifier.
+
+        Will assign the headers and payload fields in the class if the format is valid and the 
+        signature is correct, and raise an exception if not.
+
+        Args:
+            sl_response: REQUIRED. A base64-encoded status_list response, acquired (eg.) from 
+            establish_connection().
+
+            verifier: REQUIRED. A callable that verifies the signature of a payload.
+
+            min_list_length: OPTIONAL. The minimum list length, recommended to be 131,072 (see S. 6.1)
+        """
+
+        if b"." in sl_response:
+            # Enveloping proof
+
+            # Check that message is in valid JWT format 
+            headers_bytes, payload_bytes, signature = sl_response.split(b".")
+            assert headers_bytes and payload_bytes and signature
+
+            # Verify signature. verifier must be of type EnvelopingTokenVerifier
+            if not verifier(headers_bytes + b"." + payload_bytes, b64url_decode(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))
+        else:
+            # Embedding proof
+
+            # Extract data
+            self.payload = json.loads(b64url_decode(sl_response))
+
+            # TODO: Verification of signature. not sure how this works for embedded proofs
+
+        # Check values of status list against provided credential
+        credential_subject = self.payload["credentialSubject"]
+        if credential_subject["statusPurpose"] != self.credential_status["statusPurpose"]:
+            raise StatusVerificationError(
+                f"statusPurpose in credential is {self.credential_status["statusPurpose"]}, while
+                statusPurpose in status list is {credential_subject["statusPurpose"]}"
+            )
+
+        # Cache returned status list as BitArray
+        bits = self.credential_status.get("statusSize")
+        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} 
+                                        bits long, but was {self._bit_array.size} bits long instead.")
+
+    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 = self.credential_status["statusListIndex"]
+
+        status = self._bit_array[idx]
+
+        return_dict = {
+            "status": status,
+            "valid": bool(status),
+        }
+
+        # 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
+

src/token_status_list/verifier.py

@@ -52,7 +52,7 @@ def __init__(
 
         self.payload = payload
 
-        self.bit_array = bit_array
+        self._bit_array = bit_array
 
     def establish_connection(
         self, 
@@ -69,6 +69,14 @@ def establish_connection(
         assert 200 <= response.status_code < 300, f"Unable to establish connection."
         self.issuer_uri = issuer_uri
         self.encoding = status_list_format
+
+        # When establishing a new connection, clear previous cached values.
+        self.headers = None
+        self.protected_headers = None
+        self.unprotected_headers = None
+        self.payload = None
+        self._bit_array = None
+
         return response.content
 
     def jwt_verify(self, sl_response: bytes, verifier: TokenVerifier):
@@ -87,7 +95,7 @@ def jwt_verify(self, sl_response: bytes, verifier: TokenVerifier):
             signer in sign_jwt() in issuer.py.
         """
         # Ensure that the format is correct
-        assert(self.encoding != "CWT", "Please use TokenStatusListVerifier.cwt_verifier() for tokens in cwt format.")
+        assert self.encoding != "CWT", "Please use TokenStatusListVerifier.cwt_verifier() for tokens in cwt format."
         if self.encoding is None:
             self.encoding = "JWT"
 
@@ -125,6 +133,7 @@ def jwt_verify(self, sl_response: bytes, verifier: TokenVerifier):
 
         self.headers = headers
         self.payload = payload
+        self._bit_array = BitArray.load(payload["status_list"])
 
     def cwt_verify(self, token: bytes, verifier: TokenVerifier):
         """ 
@@ -152,7 +161,7 @@ def cwt_verify(self, token: bytes, verifier: TokenVerifier):
             raise ImportError("cbor extra required to use this function") from err
 
         # Ensure that the format is correct
-        assert(self.encoding != "JWT", "Please use TokenStatusListVerifier.jwt_verifier() for tokens in jwt format.")
+        assert self.encoding != "JWT", "Please use TokenStatusListVerifier.jwt_verifier() for tokens in jwt format."
         if self.encoding is None:
             self.encoding = "CWT" 
 
@@ -194,6 +203,8 @@ def cwt_verify(self, token: bytes, verifier: TokenVerifier):
         self.unprotected_headers = unprotected_headers
         self.payload = payload
 
+        self._bit_array = BitArray.load(payload[STATUS_LIST])
+
     def get_status(self, idx: int) -> int:
         """
         Returns the status of an object from the status_list in payload. 
@@ -209,14 +220,10 @@ def get_status(self, idx: int) -> int:
             The status of the requested token.
         """
 
-        assert self.encoding is not None and self.payload is not None,\
+        assert self.encoding is not None and self.payload is not None and self._bit_array is not None,\
             "Before accessing the status, please verify using jwt_verify or cwt_verify"
-
-        if self.bit_array is None:    
-            status_list = self.payload["status_list"] if self.encoding == "JWT" else self.payload[STATUS_LIST]
-            self.bit_array = BitArray.load(status_list)
 
-        return self.bit_array[idx]
+        return self._bit_array[idx]
 
 
     def serialize_verifier(self) -> dict:
@@ -266,11 +273,11 @@ def deserialize_verifier(cls, seralized_verifier: dict) -> "TokenStatusListVerif
 
         if new_verifier.encoding == "JWT":
             new_verifier.headers = seralized_verifier["headers"]
-            new_verifier.bit_array = BitArray.load(seralized_verifier["payload"]["status_list"])
+            new_verifier._bit_array = BitArray.load(seralized_verifier["payload"]["status_list"])
         elif new_verifier.encoding == "CWT":
             new_verifier.protected_headers = seralized_verifier["protected_headers"]
             new_verifier.unprotected_headers = seralized_verifier["unprotected_headers"]
-            new_verifier.bit_array = BitArray.load(seralized_verifier["payload"][STATUS_LIST])
+            new_verifier._bit_array = BitArray.load(seralized_verifier["payload"][STATUS_LIST])
         else:
             raise ValueError(f"Invalid encoding: was {seralized_verifier["encoding"]} but needs to be JWT or CWT")