register device and dpp in dlt and dpp api

device/templates/details.html

@@ -29,6 +29,9 @@
         <li class="nav-item">
           <a href="#evidences" class="nav-link" data-bs-toggle="tab" data-bs-target="#evidences">{% trans 'Evidences' %}</a>
         </li>
+        <li class="nav-item">
+          <a href="#dpps" class="nav-link" data-bs-toggle="tab" data-bs-target="#evidences">{% trans 'Dpps' %}</a>
+        </li>
         <li class="nav-item">
           <a class="nav-link" href="{% url 'device:device_web' object.id %}" target="_blank">Web</a>
         </li>
@@ -236,6 +239,22 @@
         {% endfor %}
       </div>
     </div>
+    
+    <div class="tab-pane fade" id="dpps">
+      <h5 class="card-title">{% trans 'List of dpps' %}</h5>
+      <div class="list-group col-6">
+        {% for d in dpps %}
+          <div class="list-group-item">
+            <div class="d-flex w-100 justify-content-between">
+              <small class="text-muted">{{ d.timestamp }}</small>
+            </div>
+            <p class="mb-1">
+              <a href="{% url 'evidence:details' d.uuid %}">{{ d.signature }}</a>
+            </p>
+          </div>
+        {% endfor %}
+      </div>
+    </div>
   </div>
 {% endblock %}
 

device/views.py

@@ -14,6 +14,7 @@ from django.views.generic.base import TemplateView
 from dashboard.mixins import DashboardView, Http403
 from evidence.models import Annotation
 from lot.models import LotTag
+from dpp.models import Proof
 from device.models import Device
 from device.forms import DeviceFormSet
 
@@ -103,10 +104,12 @@ class DetailsView(DashboardView, TemplateView):
         context = super().get_context_data(**kwargs)
         self.object.initial()
         lot_tags = LotTag.objects.filter(owner=self.request.user.institution)
+        dpps = Proof.objects.filter(uuid_in=self.object.uuids)
         context.update({
             'object': self.object,
             'snapshot': self.object.get_last_evidence(),
             'lot_tags': lot_tags,
+            'dpps': dpps,
         })
         return context
 

dhub/urls.py

@@ -27,4 +27,5 @@ urlpatterns = [
     path("user/", include("user.urls")),
     path("lot/", include("lot.urls")),
     path('api/', include('api.urls')),
+    path('dpp/', include('dpp.urls')),
 ]

dpp/api_dlt.py

@@ -1,38 +1,106 @@
+import time
+import logging
+
+from django.conf import settings
 from ereuseapi.methods import API
 
+from dpp.models import Proof, Dpp
+
+
+logger = logging.getLogger('django')
+
+
+# """The code of the status response of api dlt."""
+STATUS_CODE = {
+    "Success": 201,
+    "Notwork": 400
+}
+
+
+ALGORITHM = "sha3-256"
+
+
+PROOF_TYPE = {
+    'Register': 'Register',
+    'IssueDPP': 'IssueDPP',
+    'proof_of_recycling': 'proof_of_recycling',
+    'Erase': 'Erase',
+    'EWaste': 'EWaste',
+}
+
 
 def connect_api():
 
-    if not session.get('token_dlt'):
+    if not settings.get('TOKEN_DLT'):
         return
 
-    token_dlt = session.get('token_dlt')
+    token_dlt = settings.get('TOKEN_DLT')
-    api_dlt = app.config.get('API_DLT')
+    api_dlt = settings.get('API_DLT')
 
     return API(api_dlt, token_dlt, "ethereum")
 
-def register_dlt():
+
-    api = self.connect_api()
+def register_dlt(chid, phid, proof_type=None):
+    api = connect_api()
     if not api:
         return
 
-    snapshot = [x for x in self.actions if x.t == 'Snapshot']
+    if proof_type:
-    if not snapshot:
+        return api.generate_proof(
+            chid,
+            ALGORITHM,
+            phid,
+            proof_type,
+            settings.get('ID_FEDERATED')
+        )
+
+    return api.register_device(
+        chid,
+        ALGORITHM,
+        phid,
+        settings.get('ID_FEDERATED')
+    )
+
+
+def issuer_dpp_dlt(dpp):
+    phid = dpp.split(":")[0]
+    api = connect_api()
+    if not api:
         return
-    snapshot = snapshot[0]
+
-    from ereuse_devicehub.modules.dpp.models import ALGORITHM
+    return api.issue_passport(
-    from ereuse_devicehub.resources.enums import StatusCode
+        dpp,
+        ALGORITHM,
+        phid,
+        settings.get('ID_FEDERATED')
+    )
+
+
+
+def save_proof(signature, ev_uuid, result, proof_type, user):
+    if result['Status'] == STATUS_CODE.get("Success"):
+        timestamp = result.get('Data', {}).get('data', {}).get('timestamp')
+
+        if not timestamp:
+            return
+
+        d = {
+            "type": proof_type,
+            "timestamp": timestamp,
+            "issuer": user.institution.id,
+            "user": user,
+            "uuid": ev_uuid,
+            "signature": signature,
+        }
+        Proof.objects.create(**d)
+
+
+def register_device_dlt(chid, phid, ev_uuid, user):
     cny_a = 1
     while cny_a:
-        api = self.connect_api()
+        result = register_dlt(chid, phid)
-        result = api.register_device(
-            self.chid,
-            ALGORITHM,
-            snapshot.phid_dpp,
-            app.config.get('ID_FEDERATED')
-        )
         try:
-            assert result['Status'] == StatusCode.Success.value
+            assert result['Status'] == STATUS_CODE.get("Success")
             assert result['Data']['data']['timestamp']
             cny_a = 0
         except Exception:
@@ -42,55 +110,42 @@ def register_dlt():
             else:
                 cny_a = 0
 
+    save_proof(phid, ev_uuid, result, PROOF_TYPE['Register'], user)
 
-    register_proof(result)
 
-    if app.config.get('ID_FEDERATED'):
+    # TODO is neccesary?
-        cny = 1
+    # if settings.get('ID_FEDERATED'):
-        while cny:
+    #     cny = 1
-            try:
+    #     while cny:
-                api.add_service(
+    #         try:
-                    self.chid,
+    #             api.add_service(
-                    'DeviceHub',
+    #                 chid,
-                    app.config.get('ID_FEDERATED'),
+    #                 'DeviceHub',
-                    'Inventory service',
+    #                 settings.get('ID_FEDERATED'),
-                    'Inv',
+    #                 'Inventory service',
-                )
+    #                 'Inv',
-                cny = 0
+    #             )
-            except Exception:
+    #             cny = 0
-                time.sleep(10)
+    #         except Exception:
+    #             time.sleep(10)
 
-def register_proof(self, result):
-    from ereuse_devicehub.modules.dpp.models import PROOF_ENUM, Proof
-    from ereuse_devicehub.resources.enums import StatusCode
 
-    if result['Status'] == StatusCode.Success.value:
+def register_passport_dlt(chid, phid, ev_uuid, user):
-        timestamp = result.get('Data', {}).get('data', {}).get('timestamp')
+    dpp = "{chid}:{phid}".format(chid=chid, phid=phid)
-
+    if Proof.objects.filter(signature=dpp, type=PROOF_TYPE['IssueDPP']).exists():
-        if not timestamp:
-            return
-
-        snapshot = [x for x in self.actions if x.t == 'Snapshot']
-        if not snapshot:
-            return
-        snapshot = snapshot[0]
-
-        d = {
-            "type": PROOF_ENUM['Register'],
-            "device": self,
-            "action": snapshot,
-            "timestamp": timestamp,
-            "issuer_id": g.user.id,
-            "documentId": snapshot.id,
-            "documentSignature": snapshot.phid_dpp,
-            "normalizeDoc": snapshot.json_hw,
-        }
-        proof = Proof(**d)
-        db.session.add(proof)
-
-    if not hasattr(self, 'components'):
         return
 
-    for c in self.components:
+    cny_a = 1
-        if isinstance(c, DataStorage):
+    while cny_a:
-            c.register_dlt()
+        try:
+            result = issuer_dpp_dlt(dpp)
+            cny_a = 0
+        except Exception as err:
+            logger.error("ERROR API issue passport return: %s", err)
+            time.sleep(10)
+
+    if result['Status'] is not STATUS_CODE.get("Success"):
+        logger.error("ERROR API issue passport return: %s", result)
+        return
+    
+    save_proof(phid, ev_uuid, result, PROOF_TYPE['IssueDPP'], user)

dpp/migrations/0001_initial.py

@@ -1,4 +1,4 @@
-# Generated by Django 5.0.6 on 2024-11-15 18:55
+# Generated by Django 5.0.6 on 2024-11-18 14:29
 
 import django.db.models.deletion
 from django.conf import settings
@@ -15,42 +15,6 @@ class Migration(migrations.Migration):
     ]
 
     operations = [
-        migrations.CreateModel(
-            name="Dpp",
-            fields=[
-                (
-                    "id",
-                    models.BigAutoField(
-                        auto_created=True,
-                        primary_key=True,
-                        serialize=False,
-                        verbose_name="ID",
-                    ),
-                ),
-                ("timestamp", models.IntegerField()),
-                ("key", models.CharField(max_length=256)),
-                ("uuid", models.UUIDField()),
-                ("signature", models.CharField(max_length=256)),
-                ("normalizeDoc", models.TextField()),
-                ("type", models.CharField(max_length=256)),
-                (
-                    "owner",
-                    models.ForeignKey(
-                        on_delete=django.db.models.deletion.CASCADE,
-                        to="user.institution",
-                    ),
-                ),
-                (
-                    "user",
-                    models.ForeignKey(
-                        blank=True,
-                        null=True,
-                        on_delete=django.db.models.deletion.SET_NULL,
-                        to=settings.AUTH_USER_MODEL,
-                    ),
-                ),
-            ],
-        ),
         migrations.CreateModel(
             name="Proof",
             fields=[
@@ -66,11 +30,9 @@ class Migration(migrations.Migration):
                 ("timestamp", models.IntegerField()),
                 ("uuid", models.UUIDField()),
                 ("signature", models.CharField(max_length=256)),
-                ("normalizeDoc", models.TextField()),
                 ("type", models.CharField(max_length=256)),
-                ("action", models.CharField(max_length=256)),
                 (
-                    "owner",
+                    "issuer",
                     models.ForeignKey(
                         on_delete=django.db.models.deletion.CASCADE,
                         to="user.institution",

dpp/models.py

@@ -5,26 +5,11 @@ from utils.constants import STR_EXTEND_SIZE
 
 
 class Proof(models.Model):
+    ## The signature can be a phid or dpp depending of type of Proof
     timestamp = models.IntegerField()
     uuid = models.UUIDField()
     signature = models.CharField(max_length=STR_EXTEND_SIZE)
-    normalizeDoc = models.TextField()
     type = models.CharField(max_length=STR_EXTEND_SIZE)
-    action = models.CharField(max_length=STR_EXTEND_SIZE)
+    issuer = models.ForeignKey(Institution, on_delete=models.CASCADE)
-    owner = models.ForeignKey(Institution, on_delete=models.CASCADE)
     user = models.ForeignKey(
         User, on_delete=models.SET_NULL, null=True, blank=True)
-
-
-class Dpp(models.Model):
-    timestamp = models.IntegerField()
-    key = models.CharField(max_length=STR_EXTEND_SIZE)
-    uuid = models.UUIDField()
-    signature = models.CharField(max_length=STR_EXTEND_SIZE)
-    normalizeDoc = models.TextField()
-    type = models.CharField(max_length=STR_EXTEND_SIZE)
-    owner = models.ForeignKey(Institution, on_delete=models.CASCADE)
-    user = models.ForeignKey(
-        User, on_delete=models.SET_NULL, null=True, blank=True)
-
-    

dpp/urls.py

@@ -0,0 +1,8 @@
+from django.urls import path
+from dpp import views 
+
+app_name = 'dpp'
+
+urlpatterns = [
+    path("<int:proof_id>/", views.LotDashboardView.as_view(), name="proof"),
+]

dpp/views.py

@@ -1,3 +1,36 @@
-from django.shortcuts import render
+from django.views.generic.edit import View
+from django.http import JsonResponse
 
-# Create your views here.
+from evidence.xapian import search
+from dpp.models import Proof
+from dpp.api_dlt import ALGORITHM
+
+
+class ProofView(View):
+    
+    def get(self, request, *args, **kwargs):
+        timestamp = kwargs.get("proof_id")
+        proof = Proof.objects.filter(timestamp=timestamp).first()
+        if not proof:
+            return JsonResponse({}, status=404)
+
+        ev_uuid = 'uuid:"{}"'.format(proof.uuid)
+        matches = search(None, ev_uuid, limit=1)
+        if not matches or matches.size() < 1:
+            return JsonResponse({}, status=404)
+        
+        for x in matches:
+            snap = x.document.get_data()
+            
+            data = {
+                "algorithm": ALGORITHM,
+                "document": snap
+            }
+
+            d = {
+                '@context': ['https://ereuse.org/proof0.json'],
+                'data': data,
+            }
+            return JsonResponse(d, status=200)
+
+        return JsonResponse({}, status=404)

evidence/models.py

@@ -1,4 +1,5 @@
 import json
+import hashlib
 
 from dmidecode import DMIParse
 from django.db import models
@@ -59,6 +60,12 @@ class Evidence:
         if a:
             self.owner = a.owner
 
+    def get_phid(self):
+        if not self.doc:
+            self.get_doc()
+            
+        return hashlib.sha3_256(json.dumps(self.doc)).hexdigest()
+
     def get_doc(self):
         self.doc = {}
         if not self.owner:

evidence/parse.py

@@ -2,11 +2,9 @@ import json
 import hashlib
 import logging
 
-from dmidecode import DMIParse
-
 from evidence.models import Annotation
 from evidence.xapian import index
-from utils.constants import CHASSIS_DH
+from dpp.api_dlt import register_device_dlt, register_passport_dlt
 from evidence.parse_details import get_inxi_key, get_inxi
 
 logger = logging.getLogger('django')
@@ -39,6 +37,8 @@ class Build:
         self.uuid = self.json['uuid']
         self.user = user
         self.hid = None
+        self.chid = None
+        self.phid = self.get_signature(self.json)
         self.generate_chids()
 
         if check:
@@ -46,6 +46,7 @@ class Build:
 
         self.index()
         self.create_annotations()
+        self.register_device_dlt()
 
     def index(self):
         snap = json.dumps(self.evidence)
@@ -69,7 +70,8 @@ class Build:
             hid = f"{manufacturer}{model}{chassis}{serial_number}{sku}"
 
 
-        return hashlib.sha3_256(hid.encode()).hexdigest()
+        self.chid = hashlib.sha3_256(hid.encode()).hexdigest()
+        return self.chid
 
     def create_annotations(self):
         annotation = Annotation.objects.filter(
@@ -120,3 +122,10 @@ class Build:
             return f"{manufacturer}{model}{chassis}{serial_number}{sku}"
 
         return f"{manufacturer}{model}{chassis}{serial_number}{sku}{mac}"
+
+    def get_signature(self, doc):
+        return hashlib.sha3_256(json.dumps(doc).encode()).hexdigest()
+
+    def register_device_dlt(self):
+        register_device_dlt(self.chid, self.phid, self.uuid, self.user)
+        register_passport_dlt(self.chid, self.phid, self.uuid, self.user)

evidence/xapian.py

@@ -22,10 +22,14 @@ def search(institution, qs, offset=0, limit=10):
     qp.set_stemming_strategy(xapian.QueryParser.STEM_SOME)
     qp.add_prefix("uuid", "uuid")
     query = qp.parse_query(qs)
-    institution_term = "U{}".format(institution.id)
+    if institution:
-    final_query = xapian.Query(
+        institution_term = "U{}".format(institution.id)
-        xapian.Query.OP_AND, query, xapian.Query(institution_term)
+        final_query = xapian.Query(
-    )
+            xapian.Query.OP_AND, query, xapian.Query(institution_term)
+        )
+    else:
+        final_query = xapian.Query(query)
+        
     enquire = xapian.Enquire(database)
     enquire.set_query(final_query)
     matches = enquire.get_mset(offset, limit)