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]>
johnandersen777 · Dec 16, 2023 · 6e09e2b · 6e09e2b
1 parent b697db6
commit 6e09e2b
Show file tree

Hide file tree

Showing 9 changed files with 131 additions and 20 deletions.
diff --git a/docs/registration_policies.md b/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()),
             },

diff --git a/scitt_emulator/key_helper_dataclasses.py b/scitt_emulator/key_helper_dataclasses.py
@@ -0,0 +1,10 @@
+import dataclasses
+
+
+@dataclasses.dataclass
+class VerificationKey:
+    cwt: cwt.COSEKey
+    cose: pycose.keys.ec2.EC2Key
+    original: Any
+    original_content_type: str
+    original_bytes: bytes
diff --git a/scitt_emulator/key_helpers.py b/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
diff --git a/scitt_emulator/key_loader_format_did_key.py b/scitt_emulator/key_loader_format_did_key.py
@@ -15,15 +15,24 @@
 def key_loader_format_did_key(
     unverified_issuer: str,
 ) -> List[Tuple[cwt.COSEKey, pycose.keys.ec2.EC2Key]]:
+    keys = []
     jwk_keys = []
     cwt_cose_keys = []
     pycose_cose_keys = []
     cryptography_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
+
+    cryptography_keys.append(
+        VerificationKey(
+            cwt=cwt_cose_key,
+            cose=pycose_cose_key,
+            original=,
+            original_content_type=,
+            original_bytes=,
+        )
+            did_key_to_cryptography_key(unverified_issuer))
 
     for cryptography_key in cryptography_keys:
         jwk_keys.append(
@@ -40,9 +49,12 @@ def key_loader_format_did_key(
             jwk_key.export_to_pem(),
             kid=jwk_key.thumbprint(),
         )
-        cwt_cose_keys.append(cwt_cose_key)
+        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)
-        pycose_cose_keys.append((cwt_cose_key, pycose_cose_key))
+        verification_key.cose = pycose_cose_key
+
+        keys.append(verification_key)
 
-    return pycose_cose_keys
+    return keys
diff --git a/scitt_emulator/key_loader_format_url_referencing_oidc_issuer.py b/scitt_emulator/key_loader_format_url_referencing_oidc_issuer.py
@@ -13,6 +13,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/jwk+json"
 
 
 def key_loader_format_url_referencing_oidc_issuer(
@@ -59,3 +63,14 @@ def key_loader_format_url_referencing_oidc_issuer(
         pycose_cose_keys.append((cwt_cose_key, pycose_cose_key))
 
     return pycose_cose_keys
+
+
+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(),
+    }
diff --git a/scitt_emulator/key_loader_format_url_referencing_x509.py b/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 {}
diff --git a/scitt_emulator/scitt.py b/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

diff --git a/scitt_emulator/verify_statement.py b/scitt_emulator/verify_statement.py
@@ -1,6 +1,7 @@
 import os
 import itertools
 import contextlib
+import dataclasses
 import urllib.parse
 import urllib.request
 import importlib.metadata
@@ -14,6 +15,7 @@
 
 from scitt_emulator.did_helpers import did_web_to_url
 from scitt_emulator.create_statement import CWTClaims
+from scitt_emulator.key_helper_dataclasses import VerificationKey
 
 
 ENTRYPOINT_KEY_LOADERS = "scitt_emulator.verify_signature.key_loaders"
@@ -22,9 +24,7 @@
 def verify_statement(
     msg: Sign1Message,
     *,
-    key_loaders: Optional[
-        List[Callable[[str], List[Tuple[cwt.COSEKey, pycose.keys.ec2.EC2Key]]]]
-    ] = None,
+    key_loaders: Optional[List[Callable[[str], List[VerificationKey]]]] = None,
 ) -> bool:
     """
     Resolve keys for statement issuer and verify signature on COSESign1
@@ -52,15 +52,18 @@ def verify_statement(
     )
     unverified_issuer = cwt_unverified_protected[1]
 
-    # Load keys from issuer and attempt verification. Return keys used to verify
-    # as tuple of cwt.COSEKey and pycose.keys formats
-    for cwt_cose_key, pycose_cose_key in itertools.chain(
+    # Load keys from issuer and attempt verification. Return key used to verify
+    for verification_key in itertools.chain(
         *[key_loader(unverified_issuer) for key_loader in key_loaders]
     ):
-        msg.key = pycose_cose_key
+        # Skip keys that we couldn't derive COSE keys for
+        if not verification_key.cose:
+            # TODO Logging
+            continue
+        msg.key = verification_key.cose
         with contextlib.suppress(Exception):
             verify_signature = msg.verify_signature()
         if verify_signature:
-            return cwt_cose_key, pycose_cose_key
+            return verification_key
 
-    return None, None
+    return None
diff --git a/setup.py b/setup.py
@@ -17,6 +17,12 @@
             'url_referencing_ssh_authorized_keys=scitt_emulator.key_loader_format_url_referencing_ssh_authorized_keys:key_loader_format_url_referencing_ssh_authorized_keys',
             'url_referencing_x509=scitt_emulator.key_loader_format_url_referencing_x509:key_loader_format_url_referencing_x509',
         ],
+        'scitt_emulator.key_helpers.verification_key_to_object': [
+            # TODO 'to_object_did_key=scitt_emulator.key_loader_format_did_key:to_object_did_key',
+            'to_object_x509=scitt_emulator.key_loader_format_url_referencing_x509:to_object_x509',
+            # TODO 'to_object_ssh_authorized_keys=scitt_emulator.key_loader_format_url_referencing_ssh_authorized_keys:to_object_ssh_authorized_keys',
+            'to_object_oidc_issuer=scitt_emulator.key_loader_format_url_referencing_oidc_issuer:to_object_oidc_issuer',
+        ],
     },
     python_requires=">=3.8",
     install_requires=[