Fixes pretty-receipt command for COSE envelopes with embedded receipts (

#219)

.gitignore

@@ -2,6 +2,7 @@ __pycache__/
 *.pyc
 build/
 tmp/
+out/
 venv/
 .venv_ccf_sandbox/
 workspace/

pyscitt.sh

@@ -8,7 +8,7 @@ if [ ! -f "venv/bin/activate" ]; then
     echo "Setting up python virtual environment."
     python3.8 -m venv "venv"
     source venv/bin/activate 
-    pip install --disable-pip-version-check -q -e ./pyscitt
+    pip install --disable-pip-version-check --quiet --editable ./pyscitt
 else
     source venv/bin/activate 
 fi

pyscitt/pyscitt/cli/prefix_tree.py

@@ -9,7 +9,7 @@
 
 from .. import prefix_tree
 from ..client import Client
-from ..crypto import COSE_HEADER_PARAM_FEED, COSE_HEADER_PARAM_ISSUER
+from ..crypto import SCITTFeed, SCITTIssuer
 from ..verify import StaticTrustStore
 from .client_arguments import add_client_arguments, create_client
 
@@ -35,8 +35,8 @@ def prefix_tree_get_receipt(
     if claim_path:
         claim = CoseMessage.decode(claim_path.read_bytes())
 
-        issuer = claim.phdr[COSE_HEADER_PARAM_ISSUER]
-        feed = claim.phdr[COSE_HEADER_PARAM_FEED]
+        issuer = claim.get_attr(SCITTIssuer)
+        feed = claim.get_attr(SCITTFeed)
     elif not issuer or not feed:
         raise ValueError("Either a claim or an issuer and feed must be specified.")
 

pyscitt/pyscitt/cli/pretty_receipt.py

@@ -1,22 +1,49 @@
 # Copyright (c) Microsoft Corporation.
 # Licensed under the MIT License.
 import argparse
+import base64
 import json
 from pathlib import Path
+from typing import Union
 
-from ..receipt import Receipt
+import cbor2
+from pycose.messages import Sign1Message
+
+from ..receipt import Receipt, cbor_to_printable
 
 
 def prettyprint_receipt(receipt_path: Path):
+    """Pretty-print a SCITT receipt file and detect both embedded COSE_Sign1 and standalone receipt formats"""
     with open(receipt_path, "rb") as f:
         receipt = f.read()
-    parsed = Receipt.decode(receipt)
-    print(json.dumps(parsed.as_dict(), indent=2))
+
+    parsed: Union[Sign1Message, Receipt]
+    cbor_obj = cbor2.loads(receipt)
+    if hasattr(cbor_obj, "tag"):
+        assert cbor_obj.tag == 18  # COSE_Sign1
+        parsed = Sign1Message.from_cose_obj(cbor_obj.value, True)
+        output_dict = {
+            "protected": cbor_to_printable(parsed.phdr),
+            "unprotected": cbor_to_printable(parsed.uhdr),
+            "payload": (
+                base64.b64encode(parsed.payload).decode("ascii")
+                if parsed.payload
+                else None
+            ),
+        }
+    else:
+        parsed = Receipt.decode(receipt)
+        output_dict = parsed.as_dict()
+
+    fallback_serialization = lambda o: f"<<non-serializable: {type(o).__qualname__}>>"
+    print(json.dumps(output_dict, default=fallback_serialization, indent=2))
 
 
 def cli(fn):
     parser = fn(description="Pretty-print a SCITT receipt")
-    parser.add_argument("receipt", type=Path, help="Path to SCITT receipt file")
+    parser.add_argument(
+        "receipt", type=Path, help="Path to SCITT receipt file (embedded or standalone)"
+    )
 
     def cmd(args):
         prettyprint_receipt(args.receipt)

pyscitt/pyscitt/crypto.py

@@ -44,29 +44,59 @@
 )
 from cryptography.x509 import load_der_x509_certificate, load_pem_x509_certificate
 from cryptography.x509.oid import NameOID
+from pycose.headers import CoseHeaderAttribute
 from pycose.keys.cosekey import CoseKey
 from pycose.keys.curves import P256, P384, P521
 from pycose.messages import Sign1Message
 
 RECOMMENDED_RSA_PUBLIC_EXPONENT = 65537
 
 Pem = str
+RegistrationInfoValue = Union[str, bytes, int]
+RegistrationInfo = Dict[str, RegistrationInfoValue]
+CoseCurveTypes = Union[Type[P256], Type[P384], Type[P521]]
+CoseCurveType = Tuple[str, CoseCurveTypes]
+
+
+# Include SCITT-specific COSE header attributes to be recognized by pycose
+# Registered COSE header labels are in https://www.iana.org/assignments/cose/cose.xhtml
+# Draft SCITT-specific header labels are in https://datatracker.ietf.org/doc/draft-ietf-scitt-architecture/
+@CoseHeaderAttribute.register_attribute()
+class CWTClaims(CoseHeaderAttribute):
+    identifier = 15
+    fullname = "CWT_CLAIMS"
+
+
+@CoseHeaderAttribute.register_attribute()
+class SCITTIssuer(CoseHeaderAttribute):
+    identifier = 391
+    fullname = "SCITT_ISSUER"
+
+
+@CoseHeaderAttribute.register_attribute()
+class SCITTFeed(CoseHeaderAttribute):
+    identifier = 392
+    fullname = "SCITT_FEED"
 
-COSE_HEADER_PARAM_ISSUER = 391
-COSE_HEADER_PARAM_FEED = 392
-COSE_HEADER_PARAM_REGISTRATION_INFO = 393
-COSE_HEADER_PARAM_SCITT_RECEIPTS = 394
 
-COSE_HEADER_PARAM_CWT_CLAIMS = 15
+@CoseHeaderAttribute.register_attribute()
+class SCITTRegistrationInfo(CoseHeaderAttribute):
+    identifier = 393
+    fullname = "SCITT_REGISTRATION_INFO"
+
+
+@CoseHeaderAttribute.register_attribute()
+class SCITTReceipts(CoseHeaderAttribute):
+    identifier = 394
+    fullname = "SCITT_RECEIPTS"
+
+
+# CWT Claims (RFC9597) defined in https://www.iana.org/assignments/cwt/cwt.xhtml
 CWT_ISS = 1
 CWT_SUB = 2
 CWT_IAT = 6
-CWT_SVN = "svn"
-
-RegistrationInfoValue = Union[str, bytes, int]
-RegistrationInfo = Dict[str, RegistrationInfoValue]
-CoseCurveTypes = Union[Type[P256], Type[P384], Type[P521]]
-CoseCurveType = Tuple[str, CoseCurveTypes]
+# Other expected CWT claims
+CWT_SVN = "svn"  # AMD Security Version Number
 
 
 def ec_curve_from_name(name: str) -> EllipticCurve:
@@ -432,7 +462,7 @@ def embed_receipt_in_cose(buf: bytes, receipt: bytes) -> bytes:
     else:
         val = outer
     [_, uhdr, _, _] = val
-    key = COSE_HEADER_PARAM_SCITT_RECEIPTS
+    key = SCITTReceipts.identifier
     if key not in uhdr:
         uhdr[key] = []
     uhdr[key].append(parsed_receipt)
@@ -448,7 +478,7 @@ def get_last_embedded_receipt_from_cose(buf: bytes) -> Union[bytes, None]:
     else:
         val = outer
     [_, uhdr, _, _] = val
-    key = COSE_HEADER_PARAM_SCITT_RECEIPTS
+    key = SCITTReceipts.identifier
     if key in uhdr:
         parsed_receipts = uhdr[key]
         if isinstance(parsed_receipts, list) and parsed_receipts:
@@ -570,17 +600,17 @@ def sign_claimset(
             cwt_claims[CWT_SUB] = feed
         if svn is not None:
             cwt_claims[CWT_SVN] = svn
-        headers[COSE_HEADER_PARAM_CWT_CLAIMS] = cwt_claims
+        headers[CWTClaims] = cwt_claims
     else:
         if signer.issuer is not None:
-            headers[COSE_HEADER_PARAM_ISSUER] = signer.issuer
+            headers[SCITTIssuer] = signer.issuer
         if feed is not None:
-            headers[COSE_HEADER_PARAM_FEED] = feed
+            headers[SCITTFeed] = feed
         if svn is not None:
             headers["svn"] = svn
 
     if registration_info:
-        headers[COSE_HEADER_PARAM_REGISTRATION_INFO] = registration_info
+        headers[SCITTRegistrationInfo] = registration_info
 
     msg = Sign1Message(phdr=headers, payload=claims)
     msg.key = CoseKey.from_pem_private_key(signer.private_key)

pyscitt/pyscitt/prefix_tree.py

@@ -13,8 +13,8 @@
 if TYPE_CHECKING:
     from .client import BaseClient
 
-from .crypto import COSE_HEADER_PARAM_FEED, COSE_HEADER_PARAM_ISSUER
-from .receipt import ReceiptContents, hdr_as_dict
+from .crypto import SCITTFeed, SCITTIssuer
+from .receipt import ReceiptContents, cbor_to_printable
 from .verify import ServiceParameters
 
 
@@ -129,8 +129,8 @@ def root(self, claim: Sign1Message) -> bytes:
 
     @classmethod
     def claim_index(cls, claim: Sign1Message):
-        issuer = claim.phdr[COSE_HEADER_PARAM_ISSUER]
-        feed = claim.phdr.get(COSE_HEADER_PARAM_FEED, "")
+        issuer = claim.get_attr(SCITTIssuer)
+        feed = claim.get_attr(SCITTFeed, "")
         return hashlib.sha256(cbor2.dumps([issuer, feed])).digest()
 
     def leaf_hash(self, index: bytes, claim: Sign1Message) -> bytes:
@@ -151,8 +151,8 @@ def leaf_tbs(self, claim: Sign1Message) -> bytes:
 
     def as_dict(self) -> dict:
         return {
-            "tree_headers": hdr_as_dict(self.tree_headers),
-            "leaf_headers": hdr_as_dict(self.leaf_headers),
+            "tree_headers": cbor_to_printable(self.tree_headers),
+            "leaf_headers": cbor_to_printable(self.leaf_headers),
             "proof": {
                 "positions": self.proof.positions.hex(),
                 "hashes": [h.hex() for h in self.proof.hashes],

pyscitt/pyscitt/receipt.py

@@ -2,15 +2,18 @@
 # Licensed under the MIT License.
 
 import base64
+import datetime
 import hashlib
 from abc import ABC, abstractmethod
 from dataclasses import dataclass
-from typing import TYPE_CHECKING, Any
+from typing import TYPE_CHECKING, Any, Union
 
 import cbor2
 import ccf.receipt
+from cbor2 import CBORError
 from cryptography.hazmat.primitives.asymmetric import ec
 from cryptography.x509 import load_der_x509_certificate
+from pycose.headers import KID, X5chain, X5t
 from pycose.messages import Sign1Message
 from pycose.messages.cosebase import CoseBase
 
@@ -21,27 +24,92 @@
 
 HEADER_PARAM_TREE_ALGORITHM = "tree_alg"
 TREE_ALGORITHM_CCF = "CCF"
-
-
-def hdr_as_dict(phdr: dict) -> dict:
+COMMON_CWT_KEYS_MAP = {
+    1: "iss",
+    2: "sub",
+    3: "aud",
+    4: "exp",
+    5: "nbf",
+    6: "iat",
+    7: "cti",
+}
+
+
+def display_cwt_key(item: Any) -> Union[int, str]:
+    """Convert a CWT key to a string for pretty-printing."""
+    out = str(item)
+    return COMMON_CWT_KEYS_MAP.get(item, out)
+
+
+def display_cbor_val(item: Any) -> str:
+    """Convert a CBOR item to a string for pretty-printing."""
+    out = str(item)
+    if hasattr(item, "__name__"):
+        out = item.__name__
+    elif isinstance(item, datetime.datetime):
+        out = item.isoformat()
+    elif type(item) is bytes:
+        out = item.hex()
+    return out
+
+
+def cbor_to_printable(cbor_obj: Any, cbor_obj_key: Any = None) -> Any:
     """
-    Return a representation of a list of COSE header parameters that
-    is amenable to pretty-printing.
+    Return a printable representation of a CBOR object.
     """
 
-    def display(item):
-        if hasattr(item, "__name__"):
-            return item.__name__
-        if type(item) is bytes:
-            return item.hex()
-        return item
+    # pycose will use class instances for known and registered headers instead of ints
+    if hasattr(cbor_obj_key, "identifier"):
+        if cbor_obj_key.identifier == crypto.SCITTReceipts.identifier:
+            parsed_receipts = []
+            for item in cbor_obj:
+                if type(item) is bytes:
+                    try:
+                        receipt_as_dict = Receipt.decode(item).as_dict()
+                    except Exception:
+                        receipt_as_dict = {
+                            "error": "Failed to parse receipt",
+                            "cbor": item.hex(),
+                        }
+                else:
+                    try:
+                        receipt_as_dict = Receipt.from_cose_obj(item).as_dict()
+                    except Exception:
+                        receipt_as_dict = {
+                            "error": "Failed to parse receipt",
+                            "cbor": item,
+                        }
+                parsed_receipts.append(receipt_as_dict)
+            return parsed_receipts
+        if cbor_obj_key.identifier == crypto.CWTClaims.identifier:
+            return {
+                display_cwt_key(k): cbor_to_printable(v, k) for k, v in cbor_obj.items()
+            }
+        if cbor_obj_key.identifier == X5chain.identifier:
+            return [base64.b64encode(cert).decode("ascii") for cert in cbor_obj]
+        if cbor_obj_key.identifier == KID.identifier:
+            return cbor_obj.decode()
+        if cbor_obj_key.identifier == X5t.identifier:
+            return {"alg": cbor_obj[0], "hash": cbor_obj[1].hex()}
+
+    if isinstance(cbor_obj, list):
+        if not cbor_obj_key:
+            cbor_obj_key = "idx"
+        out_key = display_cbor_val(cbor_obj_key)
+        return {
+            display_cbor_val(f"{out_key}_{idx}"): cbor_to_printable(
+                v, f"{out_key}_{idx}"
+            )
+            for idx, v in enumerate(cbor_obj)
+        }
 
-    # Decode KID into a 'readable' text string if present.
-    hdr_dict = {display(k): display(v) for k, v in phdr.items()}
-    if hdr_dict.get("KID"):
-        hdr_dict["KID"] = bytes.fromhex(hdr_dict["KID"]).decode()
+    if isinstance(cbor_obj, dict):
+        return {
+            display_cbor_val(k): cbor_to_printable(v, k) for k, v in cbor_obj.items()
+        }
 
-    return hdr_dict
+    # otherwise return as is
+    return display_cbor_val(cbor_obj)
 
 
 @dataclass
@@ -186,6 +254,6 @@ def as_dict(self) -> dict:
         to pretty-printing.
         """
         return {
-            "protected": hdr_as_dict(self.phdr),
+            "protected": cbor_to_printable(self.phdr),
             "contents": self.contents.as_dict(),
         }