Compare commits
No commits in common. "dpp" and "main" have entirely different histories.
|
@ -29,9 +29,6 @@
|
|||
<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>
|
||||
|
@ -232,22 +229,6 @@
|
|||
{% 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 %}
|
||||
|
||||
|
|
|
@ -14,7 +14,6 @@ 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
|
||||
|
||||
|
@ -104,12 +103,10 @@ 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
|
||||
|
||||
|
|
|
@ -89,7 +89,6 @@ INSTALLED_APPS = [
|
|||
"dashboard",
|
||||
"admin",
|
||||
"api",
|
||||
"dpp",
|
||||
]
|
||||
|
||||
|
||||
|
|
|
@ -27,5 +27,4 @@ urlpatterns = [
|
|||
path("user/", include("user.urls")),
|
||||
path("lot/", include("lot.urls")),
|
||||
path('api/', include('api.urls')),
|
||||
path('dpp/', include('dpp.urls')),
|
||||
]
|
||||
|
|
|
@ -1,3 +0,0 @@
|
|||
from django.contrib import admin
|
||||
|
||||
# Register your models here.
|
151
dpp/api_dlt.py
151
dpp/api_dlt.py
|
@ -1,151 +0,0 @@
|
|||
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)
|
|
@ -1,6 +0,0 @@
|
|||
from django.apps import AppConfig
|
||||
|
||||
|
||||
class DppConfig(AppConfig):
|
||||
default_auto_field = "django.db.models.BigAutoField"
|
||||
name = "dpp"
|
|
@ -1,52 +0,0 @@
|
|||
# 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,
|
||||
),
|
||||
),
|
||||
],
|
||||
),
|
||||
]
|
|
@ -1,15 +0,0 @@
|
|||
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)
|
|
@ -1,3 +0,0 @@
|
|||
from django.test import TestCase
|
||||
|
||||
# Create your tests here.
|
|
@ -1,8 +0,0 @@
|
|||
from django.urls import path
|
||||
from dpp import views
|
||||
|
||||
app_name = 'dpp'
|
||||
|
||||
urlpatterns = [
|
||||
path("<int:proof_id>/", views.LotDashboardView.as_view(), name="proof"),
|
||||
]
|
36
dpp/views.py
36
dpp/views.py
|
@ -1,36 +0,0 @@
|
|||
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)
|
|
@ -1,5 +1,4 @@
|
|||
import json
|
||||
import hashlib
|
||||
|
||||
from dmidecode import DMIParse
|
||||
from django.db import models
|
||||
|
@ -59,12 +58,6 @@ 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:
|
||||
|
|
|
@ -8,13 +8,11 @@ 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:
|
||||
|
@ -42,8 +40,6 @@ 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:
|
||||
|
@ -51,7 +47,6 @@ class Build:
|
|||
|
||||
self.index()
|
||||
self.create_annotations()
|
||||
self.register_device_dlt()
|
||||
|
||||
def index(self):
|
||||
snap = json.dumps(self.json)
|
||||
|
@ -75,8 +70,7 @@ class Build:
|
|||
hid = f"{manufacturer}{model}{chassis}{serial_number}{sku}"
|
||||
|
||||
|
||||
self.chid = hashlib.sha3_256(hid.encode()).hexdigest()
|
||||
return self.chid
|
||||
return hashlib.sha3_256(hid.encode()).hexdigest()
|
||||
|
||||
def create_annotations(self):
|
||||
annotation = Annotation.objects.filter(
|
||||
|
@ -135,10 +129,3 @@ 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)
|
||||
|
|
|
@ -22,14 +22,10 @@ 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)
|
||||
if institution:
|
||||
institution_term = "U{}".format(institution.id)
|
||||
final_query = xapian.Query(
|
||||
xapian.Query.OP_AND, query, xapian.Query(institution_term)
|
||||
)
|
||||
else:
|
||||
final_query = xapian.Query(query)
|
||||
|
||||
institution_term = "U{}".format(institution.id)
|
||||
final_query = xapian.Query(
|
||||
xapian.Query.OP_AND, query, xapian.Query(institution_term)
|
||||
)
|
||||
enquire = xapian.Enquire(database)
|
||||
enquire.set_query(final_query)
|
||||
matches = enquire.get_mset(offset, limit)
|
||||
|
|
Loading…
Reference in New Issue