verify statement: As standalone file

Signed-off-by: John Andersen <[email protected]>

Author: pdxjohnny

docs/registration_policies.md

@@ -104,108 +104,7 @@ import cryptography.hazmat.primitives.serialization
 import jwcrypto.jwk
 
 from scitt_emulator.scitt import ClaimInvalidError, CWTClaims
-
-
-def did_web_to_url(
-    did_web_string, scheme=os.environ.get("DID_WEB_ASSUME_SCHEME", "https")
-):
-    return "/".join(
-        [
-            f"{scheme}:/",
-            *[urllib.parse.unquote(i) for i in did_web_string.split(":")[2:]],
-        ]
-    )
-
-
-def verify_signature(msg: Sign1Message) -> bool:
-    """
-    - TODOs
-        - Should we use audiance? I think no, just want to make sure we've
-          documented why thought if not. No usage makes sense to me becasue we
-          don't know the intended audiance, it could be federated into
-          multiple TS
-        - Can you just pass a whole public key as an issuer?
-        - Resolve DID keys (since that is what the arch says...)
-    """
-
-    # Figure out what the issuer is
-    cwt_cose_loads = cwt.cose.COSE()._loads
-    cwt_unverified_protected = cwt_cose_loads(
-        cwt_cose_loads(msg.phdr[CWTClaims]).value[2]
-    )
-    unverified_issuer = cwt_unverified_protected[1]
-
-    if unverified_issuer.startswith("did:web:"):
-        unverified_issuer = did_web_to_url(unverified_issuer)
-
-    # Load keys from issuer
-    jwk_keys = []
-    cwt_cose_keys = []
-    pycose_cose_keys = []
-
-    from cryptography.hazmat.primitives import serialization
-
-    cryptography_ssh_keys = []
-    if "://" in unverified_issuer and not unverified_issuer.startswith("file://"):
-        # TODO Logging for URLErrors
-        # Check if OIDC issuer
-        unverified_issuer_parsed_url = urllib.parse.urlparse(unverified_issuer)
-        openid_configuration_url = unverified_issuer_parsed_url._replace(
-            path="/.well-known/openid-configuration",
-        ).geturl()
-        with contextlib.suppress(urllib.request.URLError):
-            with urllib.request.urlopen(openid_configuration_url) as response:
-                if response.status == 200:
-                    openid_configuration = json.loads(response.read())
-                    jwks_uri = openid_configuration["jwks_uri"]
-                    with urllib.request.urlopen(jwks_uri) as response:
-                        if response.status == 200:
-                            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),
-                                )
-
-        # Try loading ssh keys. Example: https://github.com/username.keys
-        with contextlib.suppress(urllib.request.URLError):
-            with urllib.request.urlopen(unverified_issuer) as response:
-                while line := response.readline():
-                    with contextlib.suppress(
-                        (ValueError, cryptography.exceptions.UnsupportedAlgorithm)
-                    ):
-                        cryptography_ssh_keys.append(
-                            cryptography.hazmat.primitives.serialization.load_ssh_public_key(
-                                line
-                            )
-                        )
-
-    for cryptography_ssh_key in cryptography_ssh_keys:
-        jwk_keys.append(
-            jwcrypto.jwk.JWK.from_pem(
-                cryptography_ssh_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))
-
-    for cwt_cose_key, pycose_cose_key in pycose_cose_keys:
-        with contextlib.suppress(Exception):
-            msg.key = pycose_cose_key
-            verify_signature = msg.verify_signature()
-            if verify_signature:
-                return cwt_cose_key, pycose_cose_key
+from scitt_emulator.verify_statement import verify_statement
 
 
 def main():
@@ -220,7 +119,7 @@ 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_signature(msg)
+    cwt_cose_key, _pycose_cose_key = verify_statement(msg)
     unittest.TestCase().assertTrue(
         cwt_cose_key,
         "Failed to verify signature on statement",

pytest.ini

@@ -0,0 +1,2 @@
+[pytest]
+addopts = --doctest-modules

scitt_emulator/did_helpers.py

@@ -0,0 +1,68 @@
+import os
+import sys
+import inspect
+import urllib.parse
+from typing import Optional, Callable
+
+
+def did_web_to_url(
+    did_web_string: str,
+    *,
+    scheme: str = os.environ.get("DID_WEB_ASSUME_SCHEME", "https"),
+):
+    return "/".join(
+        [
+            f"{scheme}:/",
+            *[urllib.parse.unquote(i) for i in did_web_string.split(":")[2:]],
+        ]
+    )
+
+
+class DIDKeyDecoderNotFoundError(NotImplementedError):
+    """
+    Raised when we don't have a function implemented to decode the given key
+    """
+
+
+DID_KEY_METHOD = "did:key:"
+
+
+def did_key_to_jwk_dict_is_p_384_startswith_z82(did_key: str) -> dict[str, str]:
+    did_key = did_key.replace(DID_KEY_METHOD, "", 1)
+    return
+
+
+def did_key_to_jwk_dict(
+    did_key: str,
+    *,
+    decoders_by_prefix: Optional[dict[str, Callable[[str], dict[str, str]]]] = None,
+) -> dict[str, str]:
+    """
+    References
+
+    - https://w3c-ccg.github.io/did-method-key/#p-384
+    - RFC7515: JSON Web Key (JWK): https://www.rfc-editor.org/rfc/rfc7517
+    - RFC8037: CFRG Elliptic Curve Diffie-Hellman (ECDH) and Signatures in JSON Object Signing and Encryption (JOSE): https://www.rfc-editor.org/rfc/rfc8037
+
+    Examples
+
+    >>> did_key_to_jwk_dict("did:key:invalid")
+    Traceback (most recent call last):
+    DIDKeyDecoderNotFoundError: ...
+    >>> did_key_to_jwk_dict("did:key:z82LkvCwHNreneWpsgPEbV3gu1C6NFJEBg4srfJ5gdxEsMGRJUz2sG9FE42shbn2xkZJh54")
+    """
+    if decoders_by_prefix is None:
+        decoders_by_prefix = {
+            function_name.split("_")[-1]: function
+            for function_name, function in inspect.getmembers(sys.modules[__name__])
+            if (
+                function_name.startswith("did_key_to_jwk_dict_")
+                and "_startswith_" in function_name
+            )
+        }
+
+    for prefix, decoder in decoders_by_prefix.items():
+        if did_key.startswith(DID_KEY_METHOD + prefix):
+            return decoder(did_key)
+
+    raise DIDKeyDecoderNotFoundError(did_key)

scitt_emulator/key_loader_format_url_referencing_oidc_issuer.py

@@ -0,0 +1,74 @@
+import os
+import sys
+import json
+import pathlib
+import unittest
+import traceback
+import contextlib
+import urllib.parse
+import urllib.request
+import importlib.metadata
+from typing import Optional, Callable
+
+import jwt
+import cbor2
+import cwt
+import cwt.algs.ec2
+import pycose
+import pycose.keys.ec2
+from pycose.messages import Sign1Message
+from cryptography.hazmat.primitives import serialization
+
+# TODO Remove this once we have a example flow for proper key verification
+import jwcrypto.jwk
+
+from scitt_emulator.scitt import ClaimInvalidError, CWTClaims
+from scitt_emulator.did_helpers import did_web_to_url
+
+
+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 = []
+
+    cryptography_ssh_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 None, None
+
+    # TODO Logging for URLErrors
+    # Check if OIDC issuer
+    unverified_issuer_parsed_url = urllib.parse.urlparse(unverified_issuer)
+    openid_configuration_url = unverified_issuer_parsed_url._replace(
+        path="/.well-known/openid-configuration",
+    ).geturl()
+    with contextlib.suppress(urllib.request.URLError):
+        with urllib.request.urlopen(openid_configuration_url) as response:
+            if response.status == 200:
+                openid_configuration = json.loads(response.read())
+                jwks_uri = openid_configuration["jwks_uri"]
+                with urllib.request.urlopen(jwks_uri) as response:
+                    if response.status == 200:
+                        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),
+                            )
+
+    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

scitt_emulator/key_loader_format_url_referencing_ssh_authorized_keys.py

@@ -0,0 +1,74 @@
+import os
+import sys
+import json
+import pathlib
+import unittest
+import traceback
+import contextlib
+import urllib.parse
+import urllib.request
+import importlib.metadata
+from typing import Optional, Callable
+
+import jwt
+import cbor2
+import cwt
+import cwt.algs.ec2
+import pycose
+import pycose.keys.ec2
+from pycose.messages import Sign1Message
+from cryptography.hazmat.primitives import serialization
+
+# TODO Remove this once we have a example flow for proper key verification
+import jwcrypto.jwk
+
+from scitt_emulator.scitt import ClaimInvalidError, CWTClaims
+from scitt_emulator.did_helpers import did_web_to_url
+
+
+def key_loader_format_url_referencing_ssh_authorized_keys(
+    unverified_issuer: str,
+) -> list[tuple[cwt.COSEKey, pycose.keys.ec2.EC2Key]]:
+    jwk_keys = []
+    cwt_cose_keys = []
+    pycose_cose_keys = []
+
+    cryptography_ssh_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 None, None
+
+    # TODO Logging for URLErrors
+    # Check if OIDC issuer
+    unverified_issuer_parsed_url = urllib.parse.urlparse(unverified_issuer)
+    openid_configuration_url = unverified_issuer_parsed_url._replace(
+        path="/.well-known/openid-configuration",
+    ).geturl()
+    with contextlib.suppress(urllib.request.URLError):
+        with urllib.request.urlopen(openid_configuration_url) as response:
+            if response.status == 200:
+                openid_configuration = json.loads(response.read())
+                jwks_uri = openid_configuration["jwks_uri"]
+                with urllib.request.urlopen(jwks_uri) as response:
+                    if response.status == 200:
+                        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),
+                            )
+
+    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

scitt_emulator/scitt.py

@@ -14,6 +14,7 @@
 import pycose.headers
 
 from scitt_emulator.create_statement import CWTClaims
+from scitt_emulator.verify_statement import verify_statement
 
 # temporary receipt header labels, see draft-birkholz-scitt-receipts
 COSE_Headers_Service_Id = "service_id"
@@ -236,7 +237,10 @@ def _create_receipt(self, claim: bytes, entry_id: str):
         if CWTClaims not in msg.phdr:
             raise ClaimInvalidError("Claim does not have a CWTClaims header parameter")
 
-        # TODO Verify CWT
+        msg = Sign1Message.decode(claim, tag=True)
+        cwt_cose_key, _pycose_cose_key = verify_statement(msg)
+        if not cwt_cose_key:
+            raise ClaimInvalidError("Failed to verify signature on statement")
 
         # Extract fields of COSE_Sign1 for countersigning
         outer = cbor2.loads(claim)