key helpers: verification key to object: In progress

Asciinema: https://asciinema.org/a/627150
Asciinema: https://asciinema.org/a/627165
Signed-off-by: John Andersen <[email protected]>

docs/registration_policies.md

@@ -93,6 +93,7 @@ from jsonschema import validate, ValidationError
 
 from scitt_emulator.scitt import ClaimInvalidError, CWTClaims
 from scitt_emulator.verify_statement import verify_statement
+from scitt_emulator.key_helpers import verification_key_to_object
 
 
 def main():
@@ -107,23 +108,30 @@ def main():
             f"Claim content type does not start with application/json: {msg.phdr[pycose.headers.ContentType]!r}"
         )
 
-    cwt_cose_key, _pycose_cose_key = verify_statement(msg)
+    verification_key = verify_statement(msg)
     unittest.TestCase().assertTrue(
-        cwt_cose_key,
+        verification_key,
         "Failed to verify signature on statement",
     )
 
-    cwt_protected = cwt.decode(msg.phdr[CWTClaims], cwt_cose_key)
+    cwt_protected = cwt.decode(msg.phdr[CWTClaims], verification_key.cwt)
     issuer = cwt_protected[1]
     subject = cwt_protected[2]
 
+    issuer_key_as_object = verification_key_to_object(verification_key)
+    unittest.TestCase().assertTrue(
+        issuer_key_as_object,
+        "Failed to convert issuer key to JSON schema verifiable object",
+    )
+
     SCHEMA = json.loads(pathlib.Path(os.environ["SCHEMA_PATH"]).read_text())
 
     try:
         validate(
             instance={
                 "$schema": "https://schema.example.com/scitt-policy-engine-jsonschema.schema.json",
                 "issuer": issuer,
+                "issuer_key": issuer_key_as_object,
                 "subject": subject,
                 "claim": json.loads(msg.payload.decode()),
             },

scitt_emulator/key_helper_dataclasses.py

@@ -0,0 +1,18 @@
+import dataclasses
+from typing import Any
+
+import cwt
+import pycose.keys.ec2
+
+
+@dataclasses.dataclass
+class VerificationKey:
+    original: Any
+    original_content_type: str
+    original_bytes: bytes
+    original_bytes_encoding: str
+    # usable MUST only be true when the key can be used for CWT and COSESign1
+    # verification
+    usable: bool = False
+    cwt: cwt.COSEKey = None
+    cose: pycose.keys.ec2.EC2Key = None

scitt_emulator/key_helpers.py

@@ -0,0 +1,43 @@
+import itertools
+import importlib.metadata
+from typing import Optional, Callable, List, Tuple
+
+from scitt_emulator.key_helper_dataclasses import VerificationKey
+
+
+ENTRYPOINT_KEY_TRANSFORMS_TO_OBJECT = "scitt_emulator.key_helpers.verification_key_to_object"
+
+
+def verification_key_to_object(
+    verification_key: VerificationKey,
+    *,
+    key_transforms: Optional[List[Callable[[VerificationKey], dict]]] = None,
+) -> bool:
+    """
+    Resolve keys for statement issuer and verify signature on COSESign1
+    statement and embedded CWT
+    """
+    if key_transforms is None:
+        key_transforms = []
+        # There is some difference in the return value of entry_points across
+        # Python versions/envs (conda vs. non-conda). Python 3.8 returns a dict.
+        entrypoints = importlib.metadata.entry_points()
+        if isinstance(entrypoints, dict):
+            for entrypoint in entrypoints.get(ENTRYPOINT_KEY_TRANSFORMS_TO_OBJECT, []):
+                key_transforms.append(entrypoint.load())
+        elif isinstance(entrypoints, getattr(importlib.metadata, "EntryPoints", list)):
+            for entrypoint in entrypoints:
+                if entrypoint.group == ENTRYPOINT_KEY_TRANSFORMS_TO_OBJECT:
+                    key_transforms.append(entrypoint.load())
+        else:
+            raise TypeError(f"importlib.metadata.entry_points returned unknown type: {type(entrypoints)}: {entrypoints!r}")
+
+    # Load keys from issuer and attempt verification. Return key used to verify
+    for verification_key_as_object in itertools.chain(
+        *[key_transform(unverified_issuer) for key_transform in key_transforms]
+    ):
+        # Skip keys that we couldn't derive COSE keys for
+        if verification_key_as_object:
+            return verification_key_as_object
+
+    return None

scitt_emulator/key_loader_format_did_key.py

@@ -10,39 +10,51 @@
 import jwcrypto.jwk
 
 from scitt_emulator.did_helpers import DID_KEY_METHOD, did_key_to_cryptography_key
+from scitt_emulator.key_helper_dataclasses import VerificationKey
+
+
+# TODO What is the correct content type? Should we differ if it's been expanded?
+CONTENT_TYPE = "application/key+did"
 
 
 def key_loader_format_did_key(
     unverified_issuer: str,
-) -> List[Tuple[cwt.COSEKey, pycose.keys.ec2.EC2Key]]:
-    jwk_keys = []
-    cwt_cose_keys = []
-    pycose_cose_keys = []
-    cryptography_keys = []
+) -> List[VerificationKey]:
+    keys = []
 
     if not unverified_issuer.startswith(DID_KEY_METHOD):
-        return pycose_cose_keys
-
-    cryptography_keys.append(did_key_to_cryptography_key(unverified_issuer))
+        return keys
+
+    keys.append(
+        VerificationKey(
+            original=did_key_to_cryptography_key(unverified_issuer),
+            original_content_type=CONTENT_TYPE,
+            original_bytes=unverified_issuer.encode("utf-8"),
+            original_bytes_encoding="utf-8",
+        )
+    )
 
-    for cryptography_key in cryptography_keys:
-        jwk_keys.append(
-            jwcrypto.jwk.JWK.from_pem(
+    for verification_key in keys:
+        if isinstance(verification_key.original, cryptography.hazmat.primitives.asymmetric.ec.EllipticCurvePublicKey):
+            jwk_key = jwcrypto.jwk.JWK.from_pem(
                 cryptography_key.public_bytes(
                     encoding=serialization.Encoding.PEM,
                     format=serialization.PublicFormat.SubjectPublicKeyInfo,
                 )
             )
-        )
 
-    for jwk_key in jwk_keys:
-        cwt_cose_key = cwt.COSEKey.from_pem(
-            jwk_key.export_to_pem(),
-            kid=jwk_key.thumbprint(),
-        )
-        cwt_cose_keys.append(cwt_cose_key)
-        cwt_ec2_key_as_dict = cwt_cose_key.to_dict()
-        pycose_cose_key = pycose.keys.ec2.EC2Key.from_dict(cwt_ec2_key_as_dict)
-        pycose_cose_keys.append((cwt_cose_key, pycose_cose_key))
+            cwt_cose_key = cwt.COSEKey.from_pem(
+                jwk_key.export_to_pem(),
+                kid=jwk_key.thumbprint(),
+            )
+            verification_key.cwt = cwt_cose_key
+
+            cwt_ec2_key_as_dict = cwt_cose_key.to_dict()
+            pycose_cose_key = pycose.keys.ec2.EC2Key.from_dict(cwt_ec2_key_as_dict)
+            verification_key.cose = pycose_cose_key
+
+            verification_key.usable = True
+        else:
+            raise NotImplementedError(f"CWT and COSE transform from key of type {type(verification_key.original)} not implemented")
 
-    return pycose_cose_keys
+    return keys

scitt_emulator/key_loader_format_url_referencing_oidc_issuer.py

@@ -13,20 +13,75 @@
 import jwcrypto.jwk
 
 from scitt_emulator.did_helpers import did_web_to_url
+from scitt_emulator.key_helper_dataclasses import VerificationKey
+
+# TODO MOVE BEGIN
+import itertools
+import importlib.metadata
+from typing import Optional, Callable, List, Tuple
+
+from scitt_emulator.key_helper_dataclasses import VerificationKey
+
+
+ENTRYPOINT_KEY_TRANSFORMS_KEY_INSTANCES = "scitt_emulator.key_helpers.transforms_key_instances"
+
+
+def verification_key_to_object(
+    verification_key: VerificationKey,
+    *,
+    key_transforms: Optional[List[Callable[[VerificationKey], dict]]] = None,
+) -> bool:
+    """
+    Resolve keys for statement issuer and verify signature on COSESign1
+    statement and embedded CWT
+    """
+    if key_transforms is None:
+        key_transforms = []
+        # There is some difference in the return value of entry_points across
+        # Python versions/envs (conda vs. non-conda). Python 3.8 returns a dict.
+        entrypoints = importlib.metadata.entry_points()
+        if isinstance(entrypoints, dict):
+            for entrypoint in entrypoints.get(ENTRYPOINT_KEY_TRANSFORMS_KEY_INSTANCES, []):
+                key_transforms.append(entrypoint.load())
+        elif isinstance(entrypoints, getattr(importlib.metadata, "EntryPoints", list)):
+            for entrypoint in entrypoints:
+                if entrypoint.group == ENTRYPOINT_KEY_TRANSFORMS_KEY_INSTANCES:
+                    key_transforms.append(entrypoint.load())
+        else:
+            raise TypeError(f"importlib.metadata.entry_points returned unknown type: {type(entrypoints)}: {entrypoints!r}")
+
+    # Load keys from issuer and attempt verification. Return key used to verify
+    for verification_key_as_object in itertools.chain(
+        *[key_transform(unverified_issuer) for key_transform in key_transforms]
+    ):
+        # Skip keys that we couldn't derive COSE keys for
+        if verification_key_as_object:
+            return verification_key_as_object
+
+    return None
+
+
+def transform_key_instance_cryptography_ecc_public_to_jwcrypto_jwk(
+    key: cryptography.hazmat.primitives.asymmetric.ec.EllipticCurvePublicKey,
+) -> jwcrypto.jwk.JWK:
+        if isinstance(verification_key.transforms[-1], cryptography.hazmat.primitives.asymmetric.ec.EllipticCurvePublicKey):
+            cryptography_key = verification_key.transforms[-1]
+# TODO MOVE END
+
+
+CONTENT_TYPE = "application/jwk+json"
 
 
 def key_loader_format_url_referencing_oidc_issuer(
     unverified_issuer: str,
 ) -> List[Tuple[cwt.COSEKey, pycose.keys.ec2.EC2Key]]:
-    jwk_keys = []
-    cwt_cose_keys = []
-    pycose_cose_keys = []
+    keys = []
 
     if unverified_issuer.startswith("did:web:"):
         unverified_issuer = did_web_to_url(unverified_issuer)
 
     if "://" not in unverified_issuer or unverified_issuer.startswith("file://"):
-        return pycose_cose_keys
+        return keys
 
     # TODO Logging for URLErrors
     # Check if OIDC issuer
@@ -44,18 +99,80 @@ def key_loader_format_url_referencing_oidc_issuer(
                         jwks = json.loads(response.read())
                         for jwk_key_as_dict in jwks["keys"]:
                             jwk_key_as_string = json.dumps(jwk_key_as_dict)
-                            jwk_keys.append(
-                                jwcrypto.jwk.JWK.from_json(jwk_key_as_string),
+                            jwk_key = jwcrypto.jwk.JWK.from_json(jwk_key_as_string)
+                            keys.append(
+                                VerificationKey(
+                                    transforms=[jwk_key],
+                                    original=jwk_key,
+                                    original_content_type=CONTENT_TYPE,
+                                    original_bytes=jwk_key_as_string.encode("utf-8"),
+                                    original_bytes_encoding="utf-8",
+                                )
                             )
 
-    for jwk_key in jwk_keys:
-        cwt_cose_key = cwt.COSEKey.from_pem(
-            jwk_key.export_to_pem(),
-            kid=jwk_key.thumbprint(),
-        )
-        cwt_cose_keys.append(cwt_cose_key)
-        cwt_ec2_key_as_dict = cwt_cose_key.to_dict()
-        pycose_cose_key = pycose.keys.ec2.EC2Key.from_dict(cwt_ec2_key_as_dict)
-        pycose_cose_keys.append((cwt_cose_key, pycose_cose_key))
-
-    return pycose_cose_keys
+    for verification_key in keys:
+        for key_transformer in key_transformer:
+        if isinstance(verification_key.transforms[-1], cryptography.hazmat.primitives.asymmetric.ec.EllipticCurvePublicKey):
+            cryptography_key = verification_key.transforms[-1]
+            verification_key.transforms.append(
+                jwcrypto.jwk.JWK.from_pem(
+                    cryptography_key.public_bytes(
+                        encoding=serialization.Encoding.PEM,
+                        format=serialization.PublicFormat.SubjectPublicKeyInfo,
+                    )
+                )
+            )
+        if isinstance(verification_key.transforms[-1], jwcrypto.jwk.JWK):
+            jwk_key = verification_key.transforms[-1]
+            verification_key.transforms.append(
+                cwt.COSEKey.from_pem(
+                    jwk_key.export_to_pem(),
+                    kid=jwk_key.thumbprint(),
+                )
+            )
+        if isinstance(verification_key.transforms[-1], cwt.COSEKey):
+            cwt_cose_key = verification_key.transforms[-1]
+            cwt_ec2_key_as_dict = cwt_cose_key.to_dict()
+            verification_key.transforms.append(
+                pycose.keys.ec2.EC2Key.from_dict(cwt_ec2_key_as_dict)
+            )
+
+    for verification_key in keys:
+        for key in reversed(verification_key.transforms):
+            if not verification_key.cwt and isinstance(key, cwt.COSEKey):
+                verification_key.cwt = key
+            if (
+                not verification_key.cose
+                and isinstance(
+                    key,
+                    (
+                        pycose.keys.ec2.EC2Key,
+                    )
+                )
+            ):
+                verification_key.cose = key
+        if verification_key.cwt and verification_key.cose:
+            verification_key.usable = True
+        else:
+            raise NotImplementedError(f"CWT and COSE transform from key of type {type(verification_key.original)} not implemented")
+
+    return keys
+
+    if True:
+        key = verification_key.transforms[-1]
+        for key_transformer in key_transformer:
+        found_transformer = False
+        if verification_key.usable:
+        if not found_transformer:
+            raise NotImplementedError(f"CWT and COSE transform from key of type {type(verification_key.original)} not implemented")
+
+
+def to_object_oidc_issuer(verification_key: VerificationKey) -> dict:
+    if verification_key.original_content_type != CONTENT_TYPE:
+        return
+
+    return {
+        **verification_key.original.export_public(as_dict=True),
+        "use": "sig",
+        "kid": verification_key.original.thumbprint(),
+    }

scitt_emulator/key_loader_format_url_referencing_x509.py

@@ -14,6 +14,10 @@
 import jwcrypto.jwk
 
 from scitt_emulator.did_helpers import did_web_to_url
+from scitt_emulator.key_helper_dataclasses import VerificationKey
+
+
+CONTENT_TYPE = "application/pkix-cert"
 
 
 def key_loader_format_url_referencing_x509(
@@ -63,3 +67,13 @@ def key_loader_format_url_referencing_x509(
         pycose_cose_keys.append((cwt_cose_key, pycose_cose_key))
 
     return pycose_cose_keys
+
+
+def to_object_x509(verification_key: VerificationKey) -> dict:
+    if verification_key.original_content_type != CONTENT_TYPE:
+        return
+
+    # TODO to dict
+    verification_key.original
+
+    return {}

scitt_emulator/scitt.py

@@ -237,10 +237,10 @@ def _create_receipt(self, claim: bytes, entry_id: str):
             raise ClaimInvalidError("Claim does not have a CWTClaims header parameter")
 
         try:
-            cwt_cose_key, _pycose_cose_key = verify_statement(msg)
+            verification_key = verify_statement(msg)
         except Exception as e:
             raise ClaimInvalidError("Failed to verify signature on statement") from e
-        if not cwt_cose_key:
+        if not verification_key:
             raise ClaimInvalidError("Failed to verify signature on statement")
 
         # Extract fields of COSE_Sign1 for countersigning