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>
@@ -229,6 +232,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/settings.py

@@ -89,6 +89,7 @@ INSTALLED_APPS = [
     "dashboard",
     "admin",
     "api",
+    "dpp",
 ]
 
 

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/admin.py

@@ -0,0 +1,3 @@
+from django.contrib import admin
+
+# Register your models here.

dpp/api_dlt.py

@@ -0,0 +1,151 @@
+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 settings.get('TOKEN_DLT'):
+        return
+
+    token_dlt = settings.get('TOKEN_DLT')
+    api_dlt = settings.get('API_DLT')
+
+    return API(api_dlt, token_dlt, "ethereum")
+
+
+def register_dlt(chid, phid, proof_type=None):
+    api = connect_api()
+    if not api:
+        return
+
+    if proof_type:
+        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
+
+    return api.issue_passport(
+        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:
+        result = register_dlt(chid, phid)
+        try:
+            assert result['Status'] == STATUS_CODE.get("Success")
+            assert result['Data']['data']['timestamp']
+            cny_a = 0
+        except Exception:
+            if result.get("Data") != "Device already exists":
+                logger.error("API return: %s", result)
+                time.sleep(10)
+            else:
+                cny_a = 0
+
+    save_proof(phid, ev_uuid, result, PROOF_TYPE['Register'], user)
+
+
+    # TODO is neccesary?
+    # if settings.get('ID_FEDERATED'):
+    #     cny = 1
+    #     while cny:
+    #         try:
+    #             api.add_service(
+    #                 chid,
+    #                 'DeviceHub',
+    #                 settings.get('ID_FEDERATED'),
+    #                 'Inventory service',
+    #                 'Inv',
+    #             )
+    #             cny = 0
+    #         except Exception:
+    #             time.sleep(10)
+
+
+def register_passport_dlt(chid, phid, ev_uuid, user):
+    dpp = "{chid}:{phid}".format(chid=chid, phid=phid)
+    if Proof.objects.filter(signature=dpp, type=PROOF_TYPE['IssueDPP']).exists():
+        return
+
+    cny_a = 1
+    while cny_a:
+        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/apps.py

@@ -0,0 +1,6 @@
+from django.apps import AppConfig
+
+
+class DppConfig(AppConfig):
+    default_auto_field = "django.db.models.BigAutoField"
+    name = "dpp"

dpp/migrations/0001_initial.py

@@ -0,0 +1,52 @@
+# Generated by Django 5.0.6 on 2024-11-18 14:29
+
+import django.db.models.deletion
+from django.conf import settings
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    initial = True
+
+    dependencies = [
+        ("user", "0001_initial"),
+        migrations.swappable_dependency(settings.AUTH_USER_MODEL),
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name="Proof",
+            fields=[
+                (
+                    "id",
+                    models.BigAutoField(
+                        auto_created=True,
+                        primary_key=True,
+                        serialize=False,
+                        verbose_name="ID",
+                    ),
+                ),
+                ("timestamp", models.IntegerField()),
+                ("uuid", models.UUIDField()),
+                ("signature", models.CharField(max_length=256)),
+                ("type", models.CharField(max_length=256)),
+                (
+                    "issuer",
+                    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,
+                    ),
+                ),
+            ],
+        ),
+    ]

dpp/models.py

@@ -0,0 +1,15 @@
+from django.db import models
+from user.models import User, Institution
+from utils.constants import STR_EXTEND_SIZE
+# Create your models here.
+
+
+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)
+    type = models.CharField(max_length=STR_EXTEND_SIZE)
+    issuer = models.ForeignKey(Institution, on_delete=models.CASCADE)
+    user = models.ForeignKey(
+        User, on_delete=models.SET_NULL, null=True, blank=True)

dpp/tests.py

@@ -0,0 +1,3 @@
+from django.test import TestCase
+
+# Create your tests here.

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

@@ -0,0 +1,36 @@
+from django.views.generic.edit import View
+from django.http import JsonResponse
+
+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
@@ -58,6 +59,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

@@ -8,11 +8,13 @@ from evidence.parse_details import get_lshw_child
 
 from evidence.models import Annotation
 from evidence.xapian import index
+from dpp.api_dlt import register_device_dlt, register_passport_dlt
 from utils.constants import CHASSIS_DH
 
 
 logger = logging.getLogger('django')
 
+
 def get_mac(lshw):
     try:
         if type(lshw) is dict:
@@ -40,6 +42,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:
@@ -47,6 +51,7 @@ class Build:
 
         self.index()
         self.create_annotations()
+        self.register_device_dlt()
 
     def index(self):
         snap = json.dumps(self.json)
@@ -70,7 +75,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(
@@ -129,3 +135,10 @@ class Build:
             logger.warning(txt, snapshot['uuid'])
 
         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)