70 support did in generated gaia x credentials

.gitignore

@@ -1,5 +1,7 @@
 .idea
 .coverage
+.gx-credentials
+node_modules
 venv/
 __pycache__/
 devops/logs/

cli.py

@@ -17,8 +17,8 @@
 import openstack as os
 import yaml
 
-import generator.common.json_ld as json_ld
-from generator.common.config import Config
+import generator as json_ld
+from generator import Config
 from generator.discovery.openstack.openstack_discovery import OsCloud
 
 SHAPES_FILE_FORMAT = "turtle"

config.yaml

@@ -0,0 +1,4 @@
+issuer: "did:web:cloudandheat.com"
+pub_key:
+  - "/home/anja-strunk/Playground/gaia-x/gaia-x_ec.pem.pub"
+  - "/home/anja-strunk/Playground/gaia-x/gaia-x_rsa.pem.pub"

config/config.yaml

@@ -141,17 +141,6 @@ software resources:
     resource policy: "default: allow intent"
     license:
       - https://www.microsoft.com/windows-server/pricing
-#cloud resources:
-#  own images:
-#    AlmaLinux 8:
-#      aggregation of:
-#        - web:did:provider.de
-#      copyright owner:
-#        - "AlmaLinux OS Foundation"
-#        - "ABC"
-#      resource policy: "abc"
-#      license:
-#          - https://www.abc.org
 #wallets:
 #  filesystem:
 #    path: /etc/wallet/gx-credentials/

did_generator/cli.py

@@ -0,0 +1,48 @@
+#!/usr/bin/env python3
+
+"""Script to generate DID documents fort did:web method.
+
+(c) Anja Strunk <anja.sturnk@cloudandheat.com>, 4/2024
+SPDX-License-Identifier: EPL-2.0
+"""
+import json
+
+import click
+import yaml
+from jwcrypto.jwt import JWK
+from cryptography.hazmat.primitives import serialization
+from did_gen import DidGenerator
+from cryptography.hazmat.primitives.serialization import load_pem_public_key
+
+DEFAULT_CONFIG_FILE = "/etc/scs-did-gen/config.yaml"
+
+@click.command()
+@click.option("--config", help="Configuration file for DID generator")
+@click.option("--output-file", help="Output file - default stdout")
+def did_creator(output_file, config):
+    """Generates DID document for given DID and private keys."""
+    did_crea = DidGenerator("templates")
+
+    if not config:
+        config = DEFAULT_CONFIG_FILE
+
+    with open(config, "r") as config_file:
+        config_dict = yaml.safe_load(config_file)
+        keys = []
+        for key in config_dict['pub_key']:
+            with open(key, "rb") as key_file:
+                jwk = JWK.from_pem(load_pem_public_key(key_file.read()).public_bytes(
+                    encoding=serialization.Encoding.PEM,
+                    format=serialization.PublicFormat.SubjectPublicKeyInfo))
+                keys.append(jwk)
+
+        did_content = did_crea.generate_did_document(issuer=config_dict['issuer'], verification_methods=keys)
+        if output_file:
+            with open(output_file, "w") as did_doc:
+                did_doc.write(json.dumps(did_content, indent=4))
+        else:
+            print(json.dumps(did_content, indent=4))
+
+
+if __name__ == "__main__":
+    did_creator()

did_generator/did_gen.py

@@ -0,0 +1,44 @@
+import json
+import os
+from datetime import datetime, timezone
+from typing import Optional, List, Set
+
+import requests
+from cryptography import x509
+from cryptography.hazmat.primitives import serialization
+from dotenv import load_dotenv
+from jinja2 import Environment, FileSystemLoader, select_autoescape
+from jwcrypto import jwk
+
+# from utils import sign_doc, canonicalize, sha256_string
+
+
+class DidGenerator:
+
+    def __init__(self, jinja_templates: str):
+        self.jinja_env = Environment(
+            loader=FileSystemLoader(jinja_templates),
+            autoescape=select_autoescape()
+        )
+
+    def generate_did_document(self, issuer: str, verification_methods: List) -> dict:
+        vfy_methods = []
+        keys = []
+        key_number = 0
+        for jw_key in verification_methods:
+            jwk_content = jw_key.export(as_dict=True)
+            if jwk_content['kty'] == "RSA":
+                jwk_tmpl = self.jinja_env.get_template("rsa_jwk.j2")
+                keys.append("JWK2020-RSA-key#" + str(key_number))
+            elif jwk_content['kty'] == "EC":
+                jwk_tmpl = self.jinja_env.get_template("ec_jwk.j2")
+                keys.append("JWK2020-EC-key#" + str(key_number))
+            else:
+                raise ValueError(jwk_content['kty'] + " no supported key type.")
+
+            vfy_methods.append((jwk_tmpl.render(issuer=issuer, number=key_number, jwk=jwk_content)))
+            key_number += 1
+
+        did_doc_tmpl = self.jinja_env.get_template("did.j2")
+        did_doc = did_doc_tmpl.render(issuer=issuer, verification_method=vfy_methods, keys=keys)
+        return json.loads(did_doc)

did_generator/templates/did.j2

@@ -0,0 +1,23 @@
+{
+   "@context": [
+      "https://www.w3.org/ns/did/v1",
+      "https://w3id.org/security/suites/jws-2020/v1"
+   ],
+   "id": "{{ issuer }}",
+   "verificationMethod":[
+      {% for i in verification_method %}
+         {{ i }}
+         {%if loop.index < (keys|length) %}
+            ,
+        {% endif %}
+      {% endfor %}
+   ],
+   "assertionMethod": [
+        {% for i in keys %}
+            "{{ issuer }}{{loop.index}}#{{ i }}"
+                {%if loop.index < (keys|length) %}
+                          ,
+            {% endif %}
+        {% endfor %}
+   ]
+}

did_generator/templates/ec_jwk.j2

@@ -0,0 +1,11 @@
+    {
+      "id": "{{ issuer }}#JWK2020-EC-key#{{ number }}",
+      "type": "JsonWebKey2020",
+      "controller": "{{ issuer }}",
+      "publicKeyJwk": {
+        "kty": "EC",
+        "crv": "{{ jwk.crv }}",
+        "x": "{{ jwk.x }}",
+        "y": "{{ jwk.y }}"
+      }
+    }

did_generator/templates/rsa_jwk.j2

@@ -0,0 +1,10 @@
+      {
+         "id": "{{ issuer }}#JWK2020-RSA-key#{{ number }}",
+         "type": "JsonWebKey2020",
+         "controller": "{{ issuer }}",
+         "publicKeyJwk": {
+            "kty": "RSA",
+            "n": "{{ jwk.n }}",
+            "e": "{{ jwk.e }}"
+         }
+      }

requirements.in

@@ -5,3 +5,7 @@ click
 pyshacl
 linkml
 rdflib
+python-dotenv
+cryptography
+jwcrypto
+jinja2

requirements.txt

@@ -43,7 +43,10 @@ click==8.1.7
     #   linkml-runtime
     #   prefixcommons
 cryptography==42.0.5
-    # via openstacksdk
+    # via
+    #   -r requirements.in
+    #   jwcrypto
+    #   openstacksdk
 curies==0.7.9
     # via
     #   linkml-runtime
@@ -95,6 +98,7 @@ isoduration==20.11.0
     # via jsonschema
 jinja2==3.1.3
     # via
+    #   -r requirements.in
     #   linkml
     #   linkml-dataops
 jmespath==1.0.1
@@ -121,11 +125,12 @@ jsonpointer==2.4
     #   jsonschema
 jsonschema[format]==4.21.1
     # via
-    #   jsonschema
     #   linkml
     #   linkml-runtime
 jsonschema-specifications==2023.12.1
     # via jsonschema
+jwcrypto==1.5.6
+    # via -r requirements.in
 keystoneauth1==5.6.0
     # via openstacksdk
 kubernetes==26.1.0
@@ -219,6 +224,8 @@ python-dateutil==2.9.0.post0
     #   arrow
     #   kubernetes
     #   linkml
+python-dotenv==1.0.1
+    # via -r requirements.in
 pytrie==0.4.0
     # via curies
 pyyaml==6.0.1
@@ -314,6 +321,7 @@ types-python-dateutil==2.9.0.20240316
 typing-extensions==4.11.0
     # via
     #   dogpile-cache
+    #   jwcrypto
     #   pydantic
     #   pydantic-core
     #   sqlalchemy

tests/did_generator/test_did_gen.py

@@ -0,0 +1,50 @@
+import unittest
+
+from cryptography.hazmat.primitives.asymmetric import rsa
+from cryptography.hazmat.primitives.asymmetric import ec
+from cryptography.hazmat.primitives import serialization
+from did_generator.did_gen import DidGenerator
+from jwcrypto.jwt import JWK
+
+
+class DidGenTestCase(unittest.TestCase):
+
+    def setUp(self):
+        self.did_gen = DidGenerator("../../did_generator/templates")
+
+    def test_did_gen(self):
+        # create rsa key pair
+        private_rsa_key = rsa.generate_private_key(
+            public_exponent=3,
+            key_size=2048
+        )
+        public_rsa_key = JWK.from_pem(private_rsa_key.public_key().public_bytes(
+            encoding=serialization.Encoding.PEM,
+            format=serialization.PublicFormat.SubjectPublicKeyInfo
+        ))
+
+        # create elliptic key pair
+        private_ec_key = ec.generate_private_key(ec.SECP256R1())
+        public_ec_key = JWK.from_pem(private_ec_key.public_key().public_bytes(encoding=serialization.Encoding.PEM,
+                                                                              format=serialization.PublicFormat.SubjectPublicKeyInfo))
+
+        did_doc = self.did_gen.generate_did_document(issuer="did:web:example.com",
+                                                     verification_methods={public_rsa_key, public_ec_key})
+
+        self.assertEqual(["https://www.w3.org/ns/did/v1", "https://w3id.org/security/suites/jws-2020/v1"],
+                         did_doc['@context'])
+        self.assertEqual("did:web:example.com#JWK2020-RSA-key#0", did_doc['verificationMethod'][0]["id"])
+        self.assertEqual("JsonWebKey2020", did_doc['verificationMethod'][0]["type"])
+        self.assertEqual("did:web:example.com", did_doc['verificationMethod'][0]["controller"])
+        self.assertEqual("RSA", did_doc['verificationMethod'][0]["publicKeyJwk"]["kty"])
+        self.assertEqual("Aw", did_doc['verificationMethod'][0]["publicKeyJwk"]["e"])
+
+        self.assertEqual("did:web:example.com#JWK2020-EC-key#1", did_doc['verificationMethod'][1]["id"])
+        self.assertEqual("JsonWebKey2020", did_doc['verificationMethod'][1]["type"])
+        self.assertEqual("did:web:example.com", did_doc['verificationMethod'][1]["controller"])
+        self.assertEqual("EC", did_doc['verificationMethod'][1]["publicKeyJwk"]["kty"])
+        self.assertEqual("P-256", did_doc['verificationMethod'][1]["publicKeyJwk"]["crv"])
+
+
+if __name__ == '__main__':
+    unittest.main()

tests/test_cli.py → tests/generator/test_cli.py

@@ -8,7 +8,7 @@
 
 import cli
 from generator.common import const
-from tests.common import MockConnection, get_absolute_path
+from tests.generator.common import MockConnection, get_absolute_path
 
 OS_IMAGE_1 = OS_Image(
     hw_scsi_model="virtio - scsi",

tests/test_config.py → tests/generator/test_config.py

@@ -1,6 +1,6 @@
 import unittest
 
-import generator.common.config as config
+import generator as config
 
 
 class ConfigTestCase(unittest.TestCase):

tests/test_server_flavor_discovery.py → tests/generator/test_server_flavor_discovery.py

@@ -6,12 +6,12 @@
 from generator.common.gx_schema import CPU
 from generator.common.gx_schema import Architectures as CpuArch
 from generator.common.gx_schema import (Disk, DiskBusType, DiskType, Frequency,
-                                        Hypervisor, Memory, MemorySize)
+                       Hypervisor, Memory, MemorySize)
 from generator.common.gx_schema import ServerFlavor as GX_Flavor
 from generator.discovery.openstack.server_flavor_discovery import \
     ServerFlavorDiscovery
 from generator.vendor.flavor_names import parser_v3
-from tests.common import MockConnection, OpenstackTestcase, get_config
+from tests.generator.common import MockConnection, OpenstackTestcase, get_config
 
 OS_FLAVOR_1 = OS_Flavor(id="flavor_1", name="ABC", vcpus=2, ram=16, disk=0, description=None)
 OS_FLAVOR_2 = OS_Flavor(
@@ -149,7 +149,7 @@ def test_get_disks(self):
         gx_flavor = self.discovery._convert_to_gx(OS_Flavor(name="SCS-2C-4-10h", disk=50))
         self.assertEqual(
             [
-                Disk(diskSize=MemorySize(value=10, unit=const.UNIT_GB), diskType=DiskType("local HDD"),)
+                Disk(diskSize=MemorySize(value=10, unit=const.UNIT_GB), diskType=DiskType("local HDD"), )
             ],
             [gx_flavor.bootVolume] + gx_flavor.additionalVolume,
         )

tests/test_vm_image_discovery.py → tests/generator/test_vm_image_discovery.py

@@ -25,7 +25,7 @@
 from generator.common.gx_schema import WatchDogActions
 from generator.common.json_ld import JsonLdObject
 from generator.discovery.openstack.vm_images_discovery import VmDiscovery
-from tests.common import MockConnection, OpenstackTestcase, get_config
+from tests.generator.common import MockConnection, OpenstackTestcase, get_config
 
 GX_IMAGE_1 = JsonLdObject(
     gx_id="image_1",