Merge pull request #106 from vcstuff/awoie/add-cwt

c2bo · web-flow · commit 0da03a10b348 · 2024-03-03T22:35:51.000+01:00
Add CBOR/CWT encoding
diff --git a/.github/workflows/ghpages.yml b/.github/workflows/ghpages.yml
@@ -28,6 +28,7 @@ jobs:
 
     - name: "Generate examples"
       run: |
+        npm install -g cborg
         python3 -m pip install --upgrade pip
         python3 -m pip install -r src/requirements.txt
         python3 src/main.py
diff --git a/draft-ietf-oauth-status-list.md b/draft-ietf-oauth-status-list.md
diff --git a/example/status_list_cwt_diag b/example/status_list_cwt_diag
@@ -0,0 +1,25 @@
+d2                              # tag(18)
+  84                            #   array(4)
+    53                          #     bytes(19)
+      a20126106e7374617475736c  #       "¢\x01&\x10nstatusl"
+      6973742b637774            #       "ist+cwt"
+    a1                          #     map(1)
+      04                        #       uint(4)
+      42                        #       bytes(2)
+        3132                    #         "12"
+    58 60                       #     bytes(96)
+      a502782168747470733a2f2f  #       "¥\x02x!https://"
+      6578616d706c652e636f6d2f  #       "example.com/"
+      7374617475736c697374732f  #       "statuslists/"
+      31017368747470733a2f2f65  #       "1\x01shttps://e"
+      78616d706c652e636f6d061a  #       "xample.com\x06\x1a"
+      648c3fca041a8898c3ca19ff  #       "d\x8c?Ê\x04\x1a\x88\x98ÃÊ\x19ÿ"
+      fe56a2646269747301636c73  #       "þV¢dbits\x01cls"
+      744a78dadbb918000217015d  #       "tJxÚÛ¹\x18\x00\x02\x17\x01]"
+    58 40                       #     bytes(64)
+      3fd60a6d10eb4b4131f1f6c1  #       "?Ö\x0am\x10ëKA1ñöÁ"
+      2fb365ae27b969e8e8df0b4f  #       "/³e®'¹ièèß\x0bO"
+      4029815b679cb1051c1c9eb3  #       "@)\x81[g\x9c±\x05\x1c\x1c\x9e³"
+      6aa72f6f17bcfdb5ed443bdf  #       "j§/o\x17¼ýµíD;ß"
+      c2339568ab42949169b413e7  #       "Â3\x95h«B\x94\x91i´\x13ç"
+      02ae1e6a                  #       "\x02®\x1ej"
diff --git a/example/status_list_encoding2_cbor_diag b/example/status_list_encoding2_cbor_diag
@@ -0,0 +1,8 @@
+a2                              # map(2)
+  64                            #   string(4)
+    62697473                    #     "bits"
+  02                            #   uint(2)
+  63                            #   string(3)
+    6c7374                      #     "lst"
+  4b                            #   bytes(11)
+    78da3be9f2130003df0207      #     "xÚ;éò\x13\x00\x03ß\x02\x07"
diff --git a/example/status_list_encoding_cbor_diag b/example/status_list_encoding_cbor_diag
@@ -0,0 +1,8 @@
+a2                              # map(2)
+  64                            #   string(4)
+    62697473                    #     "bits"
+  01                            #   uint(1)
+  63                            #   string(3)
+    6c7374                      #     "lst"
+  4a                            #   bytes(10)
+    78dadbb918000217015d        #     "xÚÛ¹\x18\x00\x02\x17\x01]"
diff --git a/src/main.py b/src/main.py
@@ -50,26 +50,51 @@ def exampleStatusList2Bit() -> StatusList:
 
 def statusListEncoding1Bit():
     status_list = exampleStatusList1Bit()
-    encoded = status_list.encodeObject()
+    encoded = status_list.encodeAsJSON()
     text = "byte_array = [{}, {}] \nencoded:\n{}".format(
         hex(status_list.list[0]),
         hex(status_list.list[1]),
         util.printObject(encoded)
     )
-    util.outputFile(folder + "status_list_encoding", text)
+    util.outputFile(folder + "status_list_encoding_json", text)
 
+def statusListEncoding1BitCBOR():
+    status_list = exampleStatusList1Bit()
+    encoded = status_list.encodeAsCBOR()
+    hex_encoded = encoded.hex()
+    text = "byte_array = [{}, {}] \nencoded:\n{}".format(
+        hex(status_list.list[0]),
+        hex(status_list.list[1]),
+        util.printText(hex_encoded)
+    )
+    util.outputFile(folder + "status_list_encoding_cbor", text)
+    diag = util.printCBORDiagnostics(encoded)
+    util.outputFile(folder + "status_list_encoding_cbor_diag", diag)
 
 def statusListEncoding2Bit():
     status_list = exampleStatusList2Bit()
-    encoded = status_list.encodeObject()
+    encoded = status_list.encodeAsJSON()
     text = "byte_array = [{}, {}, {}] \nencoded:\n{}".format(
         hex(status_list.list[0]),
         hex(status_list.list[1]),
         hex(status_list.list[2]),
         util.printObject(encoded),
     )
-    util.outputFile(folder + "status_list_encoding2", text)
+    util.outputFile(folder + "status_list_encoding2_json", text)
 
+def statusListEncoding2BitCBOR():
+    status_list = exampleStatusList2Bit()
+    encoded = status_list.encodeAsCBOR()
+    hex_encoded = encoded.hex()
+    text = "byte_array = [{}, {}, {}] \nencoded:\n{}".format(
+        hex(status_list.list[0]),
+        hex(status_list.list[1]),
+        hex(status_list.list[2]),
+        util.printText(hex_encoded),
+    )
+    util.outputFile(folder + "status_list_encoding2_cbor", text)
+    diag = util.printCBORDiagnostics(encoded)
+    util.outputFile(folder + "status_list_encoding2_cbor_diag", diag)
 
 def statusListJWT():
     status_list = exampleStatusList1Bit()
@@ -83,10 +108,26 @@ def statusListJWT():
     text = util.formatToken(status_jwt, key)
     util.outputFile(folder + "status_list_jwt", text)
 
+def statusListCWT():
+    status_list = exampleStatusList1Bit()
+    cwt = StatusListToken(
+        issuer="https://example.com",
+        subject="https://example.com/statuslists/1",
+        list=status_list,
+        key=key,
+        alg=-7,
+    )
+    status_cwt = cwt.buildCWT(iat=iat, exp=exp)
+    hex_encoded = status_cwt.hex()
+    util.outputFile(folder + "status_list_cwt", util.printText(hex_encoded))
+    util.outputFile(folder + "status_list_cwt_diag", util.printCBORDiagnostics(status_cwt))
 
 if __name__ == "__main__":
     if not os.path.exists(folder):
         os.makedirs(folder)
     statusListEncoding1Bit()
     statusListEncoding2Bit()
     statusListJWT()
+    statusListEncoding1BitCBOR()
+    statusListEncoding2BitCBOR()
+    statusListCWT()
diff --git a/src/requirements.txt b/src/requirements.txt
@@ -1 +1,3 @@
 jwcrypto==1.4.2
+cbor2==5.6.1
+cwt==2.7.4
diff --git a/src/status_list.py b/src/status_list.py
@@ -1,6 +1,7 @@
 from base64 import urlsafe_b64decode, urlsafe_b64encode
 from typing import Dict
 import zlib
+from cbor2 import dumps, loads
 
 
 class StatusList:
@@ -21,18 +22,29 @@ def fromEncoded(cls, encoded: str, bits: int = 1):
         new.decode(encoded)
         return new
 
-    def encode(self) -> str:
+    def encodeAsString(self) -> str:
         zipped = zlib.compress(self.list, level=9)
         return urlsafe_b64encode(zipped).decode().strip("=")
-    
-    def encodeObject(self) -> Dict:
-        encoded_list = self.encode()
+
+    def encodeAsBytes(self) -> bytes:
+        return zlib.compress(self.list, level=9)
+
+    def encodeAsJSON(self) -> Dict:
+        encoded_list = self.encodeAsString()
         object = {
             "bits": self.bits,
             "lst": encoded_list,
         }
         return object
 
+    def encodeAsCBOR(self) -> Dict:
+        encoded_list = self.encodeAsBytes()
+        object = {
+            "bits": self.bits,
+            "lst": encoded_list,
+        }
+        return dumps(object)
+
     def decode(self, input: str):
         zipped = urlsafe_b64decode(f"{input}{'=' * divmod(len(input),4)[1]}")
         self.list = bytearray(zlib.decompress(zipped))
diff --git a/src/status_token.py b/src/status_token.py
@@ -1,11 +1,14 @@
 from jwcrypto import jwk, jwt
+from cwt import COSE, COSEKey, CWTClaims, COSEHeaders
 from status_list import StatusList
 from datetime import datetime
 from typing import Dict
+from cbor2 import dumps
 import json
 
 DEFAULT_ALG = "ES256"
-STATUS_LIST_TYP = "statuslist+jwt"
+STATUS_LIST_TYP_JWT = "statuslist+jwt"
+STATUS_LIST_TYP_CWT = "statuslist+cwt"
 
 
 class StatusListToken:
@@ -45,7 +48,7 @@ def fromJWT(cls, input: str, key: jwk.JWK):
         header = json.loads(decoded.header)
         alg = header["alg"]
         typ = header["typ"]
-        assert typ == STATUS_LIST_TYP
+        assert typ == STATUS_LIST_TYP_JWT
         claims = json.loads(decoded.claims)
         status_list = claims["status_list"]
         lst = status_list["lst"]
@@ -88,7 +91,7 @@ def buildJWT(
         claims["iat"] = int(iat.timestamp())
         if exp is not None:
             claims["exp"] = int(exp.timestamp())
-        claims["status_list"] = self.list.encodeObject()
+        claims["status_list"] = self.list.encodeAsJSON()
 
         # build header
         if optional_header is not None:
@@ -98,8 +101,57 @@ def buildJWT(
         if self._key.key_id:
             header["kid"] = self._key.key_id
         header["alg"] = self._alg
-        header["typ"] = STATUS_LIST_TYP
+        header["typ"] = STATUS_LIST_TYP_JWT
 
         token = jwt.JWT(header=header, claims=claims)
         token.make_signed_token(self._key)
         return token.serialize(compact=compact)
+    
+    def buildCWT(
+        self,
+        iat: datetime = datetime.utcnow(),
+        exp: datetime = None,
+        optional_claims: Dict = None,
+        optional_protected_header: Dict = None,
+        optional_unprotected_header: Dict = None
+    ) -> bytes:
+        # build claims
+        if optional_claims is not None:
+            claims = optional_claims
+        else:
+            claims = {}
+        claims[CWTClaims.SUB] = self.subject
+        claims[CWTClaims.ISS] = self.issuer
+        claims[CWTClaims.IAT] = int(iat.timestamp())
+        if exp is not None:
+            claims[CWTClaims.EXP] = int(exp.timestamp())
+        claims[65534] = self.list.encodeAsCBOR() # no CWT claim key assigned yet by IANA
+
+        # build header
+        if optional_protected_header is not None:
+            protected_header = optional_protected_header
+        else:
+            protected_header = {}
+
+        if optional_unprotected_header is not None:
+            unprotected_header = optional_unprotected_header
+        else:
+            unprotected_header = {}
+
+        if self._key.key_id:
+            unprotected_header[COSEHeaders.KID] = self._key.key_id.encode('utf-8')
+        protected_header[COSEHeaders.ALG] = self._alg
+        protected_header[16] = STATUS_LIST_TYP_CWT
+
+        key = COSEKey.from_jwk(self._key)
+
+        # The sender side:
+        sender = COSE.new()
+        encoded = sender.encode(
+            dumps(claims),
+            key,
+            protected=protected_header,
+            unprotected=unprotected_header
+        )
+
+        return encoded
diff --git a/src/util.py b/src/util.py
@@ -1,6 +1,7 @@
 from jwcrypto import jwk, jwt
 from textwrap import fill
 from typing import Dict
+import subprocess
 import json
 
 example = {
@@ -40,6 +41,13 @@ def printText(input: str) -> str:
     return fill(input, width=MAX_LENGTH, break_on_hyphens=False)
 
 
+# TODO: find a better way to do create CBOR Diagnostics output
+# this is still too wide
+def printCBORDiagnostics(input: bytes) -> str:
+    diag = subprocess.check_output("cborg hex2diag " + input.hex(), shell=True).decode('utf8')
+    return diag
+
+
 def outputFile(file_name: str, input: str):
     with open(file_name, "w") as file:
         file.write(input)

Original file line number	Diff line number	Diff line change
`@@ -1 +1,3 @@`
`1`	`1`	`jwcrypto==1.4.2`
	`2`	`+cbor2==5.6.1`
	`3`	`+cwt==2.7.4`