fuckin with in

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

Author: pdxjohnny

docs/registration_policies.md

@@ -203,12 +203,11 @@ for cryptography_ssh_key in cryptography_ssh_keys:
     )
 
 for jwk_key in jwk_keys:
-    print(jwk_key, "kid=", jwk_key.thumbprint())
     cwt_cose_key = cwt.COSEKey.from_pem(
         jwk_key.export_to_pem(),
-        kid=jwk_key.thumbprint(),
+        kid=jwk_key.kid,
     )
-    cwt_cose_keys.append(cwt_cose_key)
+    # cwt_cose_keys.append(cwt_cose_key)
 
 for cwt_cose_key in cwt_cose_keys:
     cwt_ec2_key_as_dict = cwt_cose_key.to_dict()
@@ -232,7 +231,7 @@ for cwt_cose_key in cwt_cose_keys:
     pprint.pprint(cwt_ec2_key_as_dict_labeled)
     pycose_cose_key = pycose.keys.ec2.EC2Key.from_dict(cwt_ec2_key_as_dict)
     pycose_cose_key.kid = cwt_ec2_key_as_dict_labeled['CRITICAL']
-    # cwt_cose_key.kid = cwt_ec2_key_as_dict_labeled['CRITICAL']
+    cwt_cose_key._kid = pycose_cose_key.kid
     pycose_cose_keys.append(pycose_cose_key)
 
 verify_signature = False
@@ -241,14 +240,17 @@ for pycose_cose_key in pycose_cose_keys:
         msg.key = pycose_cose_key
         verify_signature = msg.verify_signature()
         if verify_signature:
+            # msg.kid = pycose_cose_key.kid
             break
-            msg.kid = pycose_cose_key.kid
 
 unittest.TestCase().assertTrue(
     verify_signature,
     "Failed to verify signature on statement",
 )
 
+pprint.pprint(pycose_cose_keys)
+pprint.pprint(cwt_cose_keys)
+
 cwt_protected = cwt.decode(msg.phdr[CWTClaims], cwt_cose_keys)
 issuer = cwt_protected[1]
 subject = cwt_protected[2]

scitt_emulator/create_statement.py

@@ -83,17 +83,41 @@ def create_claim(
     if private_key_pem_path and private_key_pem_path.exists():
         cwt_cose_key = cwt.COSEKey.from_pem(private_key_pem_path.read_bytes())
     else:
-        # cwt_cose_key = cwt.COSEKey.generate_symmetric_key(alg=alg, kid=kid)
+        import subprocess
         subprocess.check_call(
             [
-                "bash"
+                "bash",
                 "-c",
                 f"ssh-keygen -q -f /dev/stdout -t ecdsa -b 384 -N '' -I {kid} <<<y 2>/dev/null | python -c 'import sys; from cryptography.hazmat.primitives import serialization; print(serialization.load_ssh_private_key(sys.stdin.buffer.read(), password=None).private_bytes(encoding=serialization.Encoding.PEM, format=serialization.PrivateFormat.PKCS8, encryption_algorithm=serialization.NoEncryption()).decode().rstrip())' > {private_key_pem_path}",
             ]
         )
         cwt_cose_key = cwt.COSEKey.from_pem(private_key_pem_path.read_bytes())
-    cwt_cose_key_to_cose_key = cwt_cose_key.to_dict()
-    sign1_message_key = pycose.keys.ec2.EC2Key.from_dict(cwt_cose_key_to_cose_key)
+    # cwt_cose_key = cwt.COSEKey.generate_ec2_key(alg=alg, kid=kid)
+    import pprint
+    cwt_ec2_key_as_dict = cwt_cose_key.to_dict()
+    pprint.pprint(cwt_ec2_key_as_dict)
+    import pprint
+    import inspect
+    cose_tags = {
+        member.identifier: member.fullname
+        for _member_name, member in inspect.getmembers(pycose.headers)
+        if (
+            hasattr(member, "identifier")
+            and hasattr(member, "fullname")
+        )
+    }
+    pprint.pprint(cose_tags)
+    cwt_ec2_key_as_dict_labeled = {
+        cose_tags.get(key, key): value
+        for key, value in cwt_ec2_key_as_dict.items()
+    }
+    # print("cwt_ec2_key_as_dict_labeled['STATIC_KEY_ID']", cwt_ec2_key_as_dict_labeled['CRITICAL'])
+    pprint.pprint(cwt_ec2_key_as_dict)
+    pprint.pprint(cwt_ec2_key_as_dict_labeled)
+    pycose_cose_key = pycose.keys.ec2.EC2Key.from_dict(cwt_ec2_key_as_dict)
+    # pycose_cose_key.kid = cwt_ec2_key_as_dict_labeled['CRITICAL']
+    # cwt_cose_key._kid = pycose_cose_key.kid
+    sign1_message_key = pycose.keys.ec2.EC2Key.from_dict(cwt_ec2_key_as_dict)
 
     # CWT_Claims (label: 14 pending [CWT_CLAIM_COSE]): A CWT representing
     # the Issuer (iss) making the statement, and the Subject (sub) to
@@ -109,11 +133,14 @@ def create_claim(
         # chosen by the Issuer
         # Example: github.com/opensbom-generator/spdx-sbom-generator/releases/tag/v0.0.13
         #   2 => tstr; sub, the subject of the statements,
-        2: "asdflkajsdflkjsadflkj" + subject,
+        2: subject,
         #   * tstr => any
     }
     # }
     cwt_token = cwt.encode(cwt_claims, cwt_cose_key)
+    print(cwt.decode(cwt_token , cwt_cose_key))
+    import sys
+    sys.exit(0)
 
     # Protected_Header = {
     protected = {

tests/test_docs.py

@@ -184,14 +184,37 @@ def test_docs_registration_policies(tmp_path):
     for name, content in docutils_find_code_samples(nodes).items():
         tmp_path.joinpath(name).write_text(content)
 
-    key = jwcrypto.jwk.JWK.generate(kty="EC", crv="P-384")
+    # key = jwcrypto.jwk.JWK.generate(kty="EC", crv="P-384")
     # cwt_cose_key = cwt.COSEKey.generate_symmetric_key(alg=alg, kid=kid)
-    private_key_pem_path.write_bytes(
-        key.export_to_pem(private_key=True, password=None),
-    )
     algorithm = "ES384"
     audience = "scitt.example.org"
     subject = "repo:scitt-community/scitt-api-emulator:ref:refs/heads/main"
+    # create claim
+    command = [
+        "client",
+        "create-claim",
+        "--out",
+        claim_path,
+        "--issuer",
+        "NOP",
+        "--subject",
+        subject,
+        "--content-type",
+        content_type,
+        "--payload",
+        payload,
+        "--private-key-pem",
+        private_key_pem_path,
+    ]
+    execute_cli(command)
+    assert os.path.exists(claim_path)
+    claim_path.unlink()
+    """
+    private_key_pem_path.write_bytes(
+        key.export_to_pem(private_key=True, password=None),
+    )
+    """
+    key = jwcrypto.jwk.JWK.from_pem(private_key_pem_path.read_bytes())
 
     # tell jsonschema_validator.py that we want to assume non-TLS URLs for tests
     os.environ["DID_WEB_ASSUME_SCHEME"] = "http"
@@ -259,6 +282,7 @@ def test_docs_registration_policies(tmp_path):
             "--url",
             service.url
         ]
+        """
         check_error = None
         try:
             execute_cli(command)
@@ -269,6 +293,7 @@ def test_docs_registration_policies(tmp_path):
         assert check_error.operation["error"] == claim_denied_error_blocked
         assert not os.path.exists(receipt_path)
         assert not os.path.exists(entry_id_path)
+        """
 
         # replace example issuer with test OIDC service issuer in allowlist
         allowlist_schema_json_path = tmp_path.joinpath("allowlist.schema.json")