Merge pull request 'DPP/DLT functionality' (#36) from dpp into main

Reviewed-on: #36
This commit is contained in:
pedro 2024-12-11 15:15:14 +00:00
commit 1ea400051c
48 changed files with 1741 additions and 53 deletions

View file

@ -3,6 +3,7 @@ DEMO=true
# note that with DEBUG=true, logs are more verbose (include tracebacks)
DEBUG=true
ALLOWED_HOSTS=localhost,localhost:8000,127.0.0.1,
DPP=false
STATIC_ROOT=/tmp/static/
MEDIA_ROOT=/tmp/media/

View file

@ -25,6 +25,7 @@ class Device:
def __init__(self, *args, **kwargs):
# the id is the chid of the device
self.id = kwargs["id"]
self.uuid = kwargs.get("uuid")
self.pk = self.id
self.shortid = self.pk[:6].upper()
self.algorithm = None
@ -103,11 +104,19 @@ class Device:
self.evidences = [Evidence(u) for u in self.uuids]
def get_last_evidence(self):
if self.last_evidence:
return
if self.uuid:
self.last_evidence = Evidence(self.uuid)
return
annotations = self.get_annotations()
if not annotations.count():
return
annotation = annotations.first()
self.last_evidence = Evidence(annotation.uuid)
self.uuid = annotation.uuid
def is_eraseserver(self):
if not self.uuids:
@ -126,6 +135,8 @@ class Device:
return False
def last_uuid(self):
if self.uuid:
return self.uuid
return self.uuids[0]
def get_lots(self):
@ -263,45 +274,38 @@ class Device:
@property
def is_websnapshot(self):
if not self.last_evidence:
self.get_last_evidence()
return self.last_evidence.doc['type'] == "WebSnapshot"
@property
def last_user_evidence(self):
if not self.last_evidence:
self.get_last_evidence()
return self.last_evidence.doc['kv'].items()
@property
def manufacturer(self):
if not self.last_evidence:
self.get_last_evidence()
return self.last_evidence.get_manufacturer()
@property
def serial_number(self):
if not self.last_evidence:
self.get_last_evidence()
return self.last_evidence.get_serial_number()
@property
def type(self):
self.get_last_evidence()
if self.last_evidence.doc['type'] == "WebSnapshot":
return self.last_evidence.doc.get("device", {}).get("type", "")
if not self.last_evidence:
self.get_last_evidence()
return self.last_evidence.get_chassis()
@property
def model(self):
if not self.last_evidence:
self.get_last_evidence()
return self.last_evidence.get_model()
@property
def components(self):
if not self.last_evidence:
self.get_last_evidence()
return self.last_evidence.get_components()

View file

@ -29,6 +29,11 @@
<li class="nav-item">
<a href="#evidences" class="nav-link" data-bs-toggle="tab" data-bs-target="#evidences">{% trans 'Evidences' %}</a>
</li>
{% if dpps %}
<li class="nav-item">
<a href="#dpps" class="nav-link" data-bs-toggle="tab" data-bs-target="#dpps">{% trans 'Dpps' %}</a>
</li>
{% endif %}
<li class="nav-item">
<a class="nav-link" href="{% url 'device:device_web' object.id %}" target="_blank">Web</a>
</li>
@ -229,6 +234,25 @@
{% endfor %}
</div>
</div>
{% if dpps %}
<div class="tab-pane fade" id="dpps">
<h5 class="card-title">{% trans 'List of dpps' %}</h5>
<div class="list-group col">
{% 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>
<span>{{ d.type }}</span>
</div>
<p class="mb-1">
<a href="{% url 'did:device_web' d.signature %}">{{ d.signature }}</a>
</p>
</div>
{% endfor %}
</div>
</div>
{% endif %}
</div>
{% endblock %}

View file

@ -1,7 +1,5 @@
import json
from django.http import JsonResponse
from django.http import Http404
from django.conf import settings
from django.urls import reverse_lazy
from django.shortcuts import get_object_or_404, Http404
from django.utils.translation import gettext_lazy as _
@ -16,6 +14,9 @@ from evidence.models import Annotation
from lot.models import LotTag
from device.models import Device
from device.forms import DeviceFormSet
if settings.DPP:
from dpp.models import Proof
from dpp.api_dlt import PROOF_TYPE
class NewDeviceView(DashboardView, FormView):
@ -103,10 +104,17 @@ class DetailsView(DashboardView, TemplateView):
context = super().get_context_data(**kwargs)
self.object.initial()
lot_tags = LotTag.objects.filter(owner=self.request.user.institution)
dpps = []
if settings.DPP:
dpps = Proof.objects.filter(
uuid__in=self.object.uuids,
type=PROOF_TYPE["IssueDPP"]
)
context.update({
'object': self.object,
'snapshot': self.object.get_last_evidence(),
'lot_tags': lot_tags,
'dpps': dpps,
})
return context

View file

@ -91,6 +91,11 @@ INSTALLED_APPS = [
"api",
]
DPP = config("DPP", default=False, cast=bool)
if DPP:
INSTALLED_APPS.extend(["dpp", "did"])
MIDDLEWARE = [
"django.middleware.security.SecurityMiddleware",
@ -239,3 +244,9 @@ LOGGING = {
SNAPSHOT_PATH="/tmp/"
DATA_UPLOAD_MAX_NUMBER_FILES = 1000
COMMIT = config('COMMIT', default='')
# DLT SETTINGS
TOKEN_DLT = config("API_DLT_TOKEN", default=None)
API_DLT = config("API_DLT", default=None)
API_RESOLVER = config("API_RESOLVER", default=None)
ID_FEDERATED = config("ID_FEDERATED", default=None)

View file

@ -14,7 +14,7 @@ Including another URLconf
1. Import the include() function: from django.urls import include, path
2. Add a URL to urlpatterns: path('blog/', include('blog.urls'))
"""
from django.conf import settings
from django.urls import path, include
urlpatterns = [
@ -28,3 +28,9 @@ urlpatterns = [
path("lot/", include("lot.urls")),
path('api/', include('api.urls')),
]
if settings.DPP:
urlpatterns.extend([
path('dpp/', include('dpp.urls')),
path('did/', include('did.urls')),
])

0
did/__init__.py Normal file
View file

3
did/admin.py Normal file
View file

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

6
did/apps.py Normal file
View file

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

View file

3
did/models.py Normal file
View file

@ -0,0 +1,3 @@
from django.db import models
# Create your models here.

View file

@ -0,0 +1,41 @@
dpp_tmpl = {
"@context": [
"https://www.w3.org/ns/credentials/v2",
"https://test.uncefact.org/vocabulary/untp/dpp/0.5.0/"
],
"type": [
"DigitalProductPassport",
"VerifiableCredential"
],
"id": "https://example.ereuse.org/credentials/2a423366-a0d6-4855-ba65-2e0c926d09b0",
"issuer": {
"type": [
"CredentialIssuer"
],
"id": "did:web:r1.identifiers.ereuse.org:did-registry:z6Mkoreij5y9bD9fL5SGW6TfMUmcbaV7LCPwZHCFEEZBrVYQ#z6Mkoreij5y9bD9fL5SGW6TfMUmcbaV7LCPwZHCFEEZBrVYQ",
"name": "Refurbisher One"
},
"validFrom": "2024-11-15T12:00:00",
"validUntil": "2034-11-15T12:00:00",
"credentialSubject": {
"type": [
"Product"
],
"id": "https://id.ereuse.org/01/09520123456788/21/12345",
"name": "Refurbished XYZ Lenovo laptop item",
"registeredId": "09520123456788.21.12345",
"description": "XYZ Lenovo laptop refurbished by Refurbisher One",
"data": ""
},
"credentialSchema": {
"id": "https://idhub.pangea.org/vc_schemas/dpp.json",
"type": "FullJsonSchemaValidator2021",
"proof": {
"type": "Ed25519Signature2018",
"proofPurpose": "assertionMethod",
"verificationMethod": "did:web:r1.identifiers.ereuse.org:did-registry:z6Mkoreij5y9bD9fL5SGW6TfMUmcbaV7LCPwZHCFEEZBrVYQ#z6Mkoreij5y9bD9fL5SGW6TfMUmcbaV7LCPwZHCFEEZBrVYQ",
"created": "2024-12-03T15:33:42Z",
"jws": "eyJhbGciOiJFZERTQSIsImNyaXQiOlsiYjY0Il0sImI2NCI6ZmFsc2V9..rBPqbOcZCXB7GAnq1XIfV9Jvw4MKXlHff7qZkRfgwQ0Hnd9Ujt5s1xT4O0K6VESzWvdP2mOvMvu780fVNfraBQ"
}
}
}

View file

@ -0,0 +1,497 @@
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8" />
<meta name="viewport" content="width=device-width, initial-scale=1.0" />
<title>{{ object.type }}</title>
<link href="https://cdnjs.cloudflare.com/ajax/libs/bootstrap/5.1.3/css/bootstrap.min.css" rel="stylesheet" />
<link rel="stylesheet" href="https://cdn.jsdelivr.net/npm/bootstrap-icons@1.11.1/font/bootstrap-icons.css" />
<style>
body {
font-size: 0.875rem;
background-color: #f8f9fa;
display: flex;
flex-direction: column;
min-height: 100vh;
}
.custom-container {
background-color: #ffffff;
border-radius: 10px;
box-shadow: 0 0 20px rgba(0, 0, 0, 0.1);
padding: 30px;
margin-top: 30px;
flex-grow: 1;
}
.section-title {
color: #7a9f4f;
border-bottom: 2px solid #9cc666;
padding-bottom: 10px;
margin-bottom: 20px;
font-size: 1.5em;
}
.info-row {
margin-bottom: 10px;
}
.info-label {
font-weight: bold;
color: #545f71;
}
.info-value {
color: #333;
}
.component-card {
background-color: #f8f9fa;
border-left: 4px solid #9cc666;
margin-bottom: 15px;
transition: all 0.3s ease;
}
.component-card:hover {
box-shadow: 0 5px 15px rgba(0, 0, 0, 0.1);
transform: translateY(-2px);
}
.hash-value {
word-break: break-all;
background-color: #f3f3f3;
padding: 5px;
border-radius: 4px;
font-family: monospace;
font-size: 0.9em;
border: 1px solid #e0e0e0;
}
.card-title {
color: #9cc666;
}
.btn-primary {
background-color: #9cc666;
border-color: #9cc666;
padding: 0.1em 2em;
font-weight: 700;
}
.btn-primary:hover {
background-color: #8ab555;
border-color: #8ab555;
}
.btn-green-user {
background-color: #c7e3a3;
}
.btn-grey {
background-color: #f3f3f3;
}
footer {
background-color: #545f71;
color: #ffffff;
text-align: center;
padding: 10px 0;
margin-top: 20px;
}
</style>
</head>
<body>
<div class="container custom-container">
<nav class="header-nav ms-auto">
<div class="d-flex align-items-right">
<span class="nav-item">
{% if not roles and user.is_anonymous %}
<button class="btn btn-primary" data-bs-toggle="modal" data-bs-target="#validateModal">Validate</button>
{% else %}
<button class="btn btn-primary" id="buttonRole" data-bs-toggle="modal" data-bs-target="#rolesModal">Select your role</button>
<a class="btn btn-primary" href="{% url 'login:logout' %}?next={{ path }}">Logout</a>
{% endif %}
</span>
</div>
{% if role %}
<div class="d-flex justify-content-end">
<span class="nav-item">
Current Role: {{ role }}
</span>
</div>
{% endif %}
</nav>
<h1 class="text-center mb-4" style="color: #545f71;">{{ object.manufacturer }} {{ object.type }} {{ object.model }}</h1>
<div class="row">
<div class="col-lg-6">
{% if manuals.details.logo %}
<img style="max-width: 50px; margin-right: 15px;" src="{{ manuals.details.logo }}" />
{% endif %}
</div>
<div class="col-lg-6">
{% if manuals.details.image %}
<img style="width: 100px;" src="{{ manuals.details.image }}" />
{% endif %}
</div>
</div>
<div class="row">
<div class="col-lg-6">
<h2 class="section-title">Details</h2>
<div class="info-row row">
<div class="col-md-4 info-label">Phid</div>
<div class="col-md-8 info-value">
<div class="hash-value">{{ object.id }}</div>
</div>
</div>
<div class="info-row row">
<div class="col-md-4 info-label">Type</div>
<div class="col-md-8 info-value">{{ object.type }}</div>
</div>
{% if object.is_websnapshot %}
{% for snapshot_key, snapshot_value in object.last_user_evidence %}
<div class="info-row row">
<div class="col-md-4 info-label">{{ snapshot_key }}</div>
<div class="col-md-8 info-value">{{ snapshot_value|default:'' }}</div>
</div>
{% endfor %}
{% else %}
<div class="info-row row">
<div class="col-md-4 info-label">Manufacturer</div>
<div class="col-md-8 info-value">{{ object.manufacturer|default:'' }}</div>
</div>
<div class="info-row row">
<div class="col-md-4 info-label">Model</div>
<div class="col-md-8 info-value">{{ object.model|default:'' }}</div>
</div>
{% if user.is_authenticated %}
<div class="info-row row">
<div class="col-md-4 info-label">Serial Number</div>
<div class="col-md-8 info-value">{{ object.serial_number|default:'' }}</div>
</div>
{% endif %}
{% endif %}
</div>
<div class="col-lg-6">
<h2 class="section-title">Identifiers</h2>
{% for chid in object.hids %}
<div class="info-row">
<div class="hash-value">{{ chid|default:'' }}</div>
</div>
{% endfor %}
</div>
</div>
<h2 class="section-title mt-5">Components</h2>
<div class="row">
{% for component in object.components %}
<div class="col-md-6 mb-3">
<div class="card component-card">
<div class="card-body">
<h5 class="card-title">{{ component.type }}</h5>
<p class="card-text">
{% for component_key, component_value in component.items %}
{% if component_key not in 'actions,type' %}
{% if component_key != 'serialNumber' or user.is_authenticated %}
<strong>{{ component_key }}:</strong> {{ component_value }}<br />
{% endif %}
{% endif %}
{% endfor %}
</p>
</div>
</div>
</div>
{% endfor %}
</div>
{% if manuals.icecat %}
<h5 class="card-title">Icecat data sheet</h5>
<div class="row">
<div class="col-12 list-group-item d-flex align-items-center">
{% if manuals.details.logo %}
<img style="max-width: 50px; margin-right: 15px;" src="{{ manuals.details.logo }}" />
{% endif %}
{% if manuals.details.image %}
<img style="max-width: 100px; margin-right: 15px;" src="{{ manuals.details.image }}" />
{% endif %}
{% if manuals.details.pdf %}
<a href="{{ manuals.details.pdf }}" target="_blank">{{ manuals.details.title }}</a><br />
{% else %}
{{ manuals.details.title }}<br />
{% endif %}
</div>
<div class="col-12 accordion-item">
<h5 class="card-title accordion-header">
<button class="accordion-button collapsed" data-bs-target="#manuals-icecat" type="button"
data-bs-toggle="collapse" aria-expanded="false">
More examples
</button>
</h5>
<div id="manuals-icecat" class="row accordion-collapse collapse">
<div class="accordion-body">
{% for m in manuals.icecat %}
<div class="list-group-item d-flex align-items-center">
{% if m.logo %}
<img style="max-width: 50px; margin-right: 15px;" src="{{ m.logo }}" />
{% endif %}
{% if m.pdf %}
<a href="{{ m.pdf }}" target="_blank">{{ m.title }}</a><br />
{% else %}
{{ m.title }}<br />
{% endif %}
</div>
{% endfor %}
</div>
</div>
</div>
</div>
{% endif %}
{% if manuals.laer %}
<div class="row mt-3">
<div class="col-12">
<h5 class="card-title">Recycled Content</h5>
<div class="row mb-3">
<div class="col-sm-2">
Metal
</div>
<div class="col-sm-10">
<div class="progress">
<div class="progress-bar"
role="progressbar"
style="width: {{ manuals.laer.0.metal }}%"
aria-valuenow="{{ manuals.laer.0.metal }}"
aria-valuemin="0"
aria-valuemax="100">{{ manuals.laer.0.metal }}%
</div>
</div>
</div>
</div>
<div class="row mb-3">
<div class="col-sm-2">
Plastic post Consumer
</div>
<div class="col-sm-10">
<div class="progress">
<div class="progress-bar"
role="progressbar"
style="width: {{ manuals.laer.0.plastic_post_consumer }}%"
aria-valuenow="{{ manuals.laer.0.plastic_post_consumer }}"
aria-valuemin="0"
aria-valuemax="100">{{ manuals.laer.0.plastic_post_consumer }}%
</div>
</div>
</div>
</div>
<div class="row mb-3">
<div class="col-sm-2">
Plastic post Industry
</div>
<div class="col-sm-10">
<div class="progress">
<div class="progress-bar"
role="progressbar"
style="width: {{ manuals.laer.0.plastic_post_industry }}%"
aria-valuenow="{{ manuals.laer.0.plastic_post_industry }}"
aria-valuemin="0"
aria-valuemax="100">{{ manuals.laer.0.plastic_post_industry }}%
</div>
</div>
</div>
</div>
</div>
</div>
{% endif %}
{% if manuals.energystar %}
<div class="row mt-3">
<div class="col-12">
<h5 class="card-title">Energy spent</h5>
{% if manuals.energystar.long_idle_watts %}
<div class="row mb-3">
<div class="col-sm-10">
Consumption when inactivity power function is activated (watts)
</div>
<div class="col-sm-2">
{{ manuals.energystar.long_idle_watts }}
</div>
</div>
{% endif %}
{% if manuals.energystar.short_idle_watts %}
<div class="row mb-3">
<div class="col-sm-10">
Consumption when inactivity power function is not activated (watts)
</div>
<div class="col-sm-2">
{{ manuals.energystar.short_idle_watts }}
</div>
</div>
{% endif %}
{% if manuals.energystar.sleep_mode_watts %}
<div class="row mb-3">
<div class="col-sm-10">
sleep_mode_watts
Consumption when computer goes into sleep mode (watts)
</div>
<div class="col-sm-2">
{{ manuals.energystar.sleep_mode_watts }}
</div>
</div>
{% endif %}
{% if manuals.energystar.off_mode_watts %}
<div class="row mb-3">
<div class="col-sm-10">
Consumption when the computer is off (watts)
</div>
<div class="col-sm-2">
{{ manuals.energystar.off_mode_watts }}
</div>
</div>
{% endif %}
{% if manuals.energystar.tec_allowance_kwh %}
<div class="row mb-3">
<div class="col-sm-10">
Power allocation for normal operation (kwh)
</div>
<div class="col-sm-2">
{{ manuals.energystar.tec_allowance_kwh }}
</div>
</div>
{% endif %}
{% if manuals.energystar.tec_of_model_kwh %}
<div class="row mb-3">
<div class="col-sm-10">
Consumption of the model configuration (kwh)
</div>
<div class="col-sm-2">
{{ manuals.energystar.tec_of_model_kwh }}
</div>
</div>
{% endif %}
{% if manuals.energystar.tec_requirement_kwh %}
<div class="row mb-3">
<div class="col-sm-10">
Energy allowance provided (kwh)
</div>
<div class="col-sm-2">
{{ manuals.energystar.tec_requirement_kwh }}
</div>
</div>
{% endif %}
{% if manuals.energystar.work_off_mode_watts %}
<div class="row mb-3">
<div class="col-sm-10">
The lowest power mode which cannot be switched off (watts)
</div>
<div class="col-sm-2">
{{ manuals.energystar.work_off_mode_watts }}
</div>
</div>
{% endif %}
{% if manuals.energystar.work_weighted_power_of_model_watts %}
<div class="row mb-3">
<div class="col-sm-10">
Weighted energy consumption from all its states (watts)
</div>
<div class="col-sm-2">
{{ manuals.energystar.work_weighted_power_of_model_watts }}
</div>
</div>
{% endif %}
</div>
</div>
{% endif %}
{% if manuals.ifixit %}
<div class="row">
<div class="col-12 accordion-item">
<h5 class="card-title accordion-header">
<button class="accordion-button collapsed" data-bs-target="#manuals-repair" type="button"
data-bs-toggle="collapse" aria-expanded="false">
Repair manuals
</button>
</h5>
<div id="manuals-repair" class="row accordion-collapse collapse">
<div class="list-group col">
{% for m in manuals.ifixit %}
<div class="list-group-item d-flex align-items-center">
{% if m.image %}
<img style="max-width: 100px; margin-right: 15px;" src="{{ m.image }}" />
{% endif %}
{% if m.url %}
<a href="{{ m.url }}" target="_blank">{{ m.title }}</a><br />
{% else %}
{{ m.title }}<br />
{% endif %}
</div>
{% endfor %}
</div>
</div>
</div>
</div>
{% endif %}
<footer>
<p>
&copy;{% now 'Y' %}eReuse. All rights reserved.
</p>
</footer>
{% if user.is_anonymous and not roles %}
<div class="modal fade" id="validateModal" tabindex="-1" style="display: none;" aria-hidden="true">
<div class="modal-dialog modal-dialog-centered">
<div class="modal-content">
<div class="modal-header">
<h5 class="modal-title">Validate as <span id="title-action"></span></h5>
<button type="button" class="btn-close" data-bs-dismiss="modal" aria-label="Close"></button>
</div>
<div class="modal-body">
<a class="btn btn-primary" type="button"
href="{% url 'login:login' %}?next={{ path }}">
User of system
</a>
{% if oidc %}
<br />
<a class="btn btn-primary mt-3" type="button" href="{# url 'oidc:login_other_inventory' #}?next={{ path }}">
User of other inventory
</a>
{% endif %}
</div>
<div class="modal-footer"></div>
</div>
</div>
</div>
{% else %}
<div class="modal fade" id="rolesModal" tabindex="-1" style="display: none;" aria-hidden="true">
<div class="modal-dialog modal-dialog-centered">
<div class="modal-content">
<form action="{{ path }}" method="get">
<div class="modal-header">
<h5 class="modal-title">Select your Role <span id="title-action"></span></h5>
<button type="button" class="btn-close" data-bs-dismiss="modal" aria-label="Close"></button>
</div>
<div class="modal-body">
<select name="role">
{% for k, v in roles %}
<option value="{{ k }}" {% if v == role %}selected=selected{% endif %}>{{ v }}</option>
{% endfor %}
</select>
</div>
<div class="modal-footer">
<button type="button" class="btn btn-secondary" data-bs-dismiss="modal">Close</button>
<input type="submit" class="btn btn-primary" value="Send" />
</div>
</form>
</div>
</div>
</div>
{% endif %}
<script src="https://cdnjs.cloudflare.com/ajax/libs/bootstrap/5.1.3/js/bootstrap.bundle.min.js"></script>
</body>
</html>

3
did/tests.py Normal file
View file

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

8
did/urls.py Normal file
View file

@ -0,0 +1,8 @@
from django.urls import path
from did import views
app_name = 'did'
urlpatterns = [
path("<str:pk>", views.PublicDeviceWebView.as_view(), name="device_web"),
]

263
did/views.py Normal file
View file

@ -0,0 +1,263 @@
import json
import logging
from django.http import JsonResponse, Http404
from django.views.generic.base import TemplateView
from device.models import Device
from evidence.parse import Build
from dpp.api_dlt import ALGORITHM
from dpp.models import Proof
from dpp.api_dlt import PROOF_TYPE
from did.template_credential import dpp_tmpl
logger = logging.getLogger('django')
class PublicDeviceWebView(TemplateView):
template_name = "device_did.html"
def get(self, request, *args, **kwargs):
self.pk = kwargs['pk']
chid = self.pk.split(":")[0]
proof = Proof.objects.filter(signature=self.pk).first()
if proof:
self.object = Device(id=chid, uuid=proof.uuid)
else:
self.object = Device(id=chid)
if not self.object.last_evidence:
raise Http404
if self.request.headers.get('Accept') == 'application/json':
return self.get_json_response()
return super().get(request, *args, **kwargs)
def get_context_data(self, **kwargs):
self.context = super().get_context_data(**kwargs)
self.object.initial()
roles = [("Operator", "Operator")]
role = "Operator"
if self.request.user.is_anonymous:
roles = []
role = None
self.context.update({
'object': self.object,
'role': role,
'roles': roles,
'path': self.request.path,
'last_dpp': "",
'before_dpp': "",
})
if not self.request.user.is_anonymous:
self.get_manuals()
return self.context
@property
def public_fields(self):
return {
'id': self.object.id,
'shortid': self.object.shortid,
'uuids': self.object.uuids,
'hids': self.object.hids,
'components': self.remove_serial_number_from(self.object.components),
}
@property
def authenticated_fields(self):
return {
'serial_number': self.object.serial_number,
'components': self.object.components,
}
def remove_serial_number_from(self, components):
for component in components:
if 'serial_number' in component:
del component['SerialNumber']
return components
def get_device_data(self):
data = self.public_fields
if self.request.user.is_authenticated:
data.update(self.authenticated_fields)
return data
def get_json_response(self):
device_data = self.get_result()
# device_data = self.get_device_data()
response = JsonResponse(device_data)
response["Access-Control-Allow-Origin"] = "*"
return response
def get_result(self):
if len(self.pk.split(":")) > 1:
return self.build_from_dpp()
else:
return self.build_from_chid()
def build_from_dpp(self):
data = {
'document': {},
'dpp': self.pk,
'algorithm': ALGORITHM,
'components': [],
'manufacturer DPP': '',
'device': {},
}
dev = Build(self.object.last_evidence.doc, None, check=True)
doc = dev.get_phid()
data['document'] = json.dumps(doc)
data['device'] = dev.device
data['components'] = dev.components
self.object.get_evidences()
last_dpp = Proof.objects.filter(
uuid__in=self.object.uuids, type=PROOF_TYPE['IssueDPP']
).order_by("-timestamp").first()
key = self.pk
if last_dpp:
key = last_dpp.signature
url = "https://{}/did/{}".format(
self.request.get_host(),
key
)
data['url_last'] = url
tmpl = dpp_tmpl.copy()
tmpl["credentialSubject"]["data"] = data
return tmpl
def build_from_chid(self):
dpps = []
self.object.initial()
for d in self.object.evidences:
d.get_doc()
dev = Build(d.doc, None, check=True)
doc = dev.get_phid()
ev = json.dumps(doc)
phid = dev.get_signature(doc)
dpp = "{}:{}".format(self.pk, phid)
rr = {
'dpp': dpp,
'document': ev,
'algorithm': ALGORITHM,
'manufacturer DPP': '',
'device': dev.device,
'components': dev.components
}
tmpl = dpp_tmpl.copy()
tmpl["credentialSubject"]["data"] = rr
dpps.append(tmpl)
return {
'@context': ['https://ereuse.org/dpp0.json'],
'data': dpps,
}
def get_manuals(self):
manuals = {
'ifixit': [],
'icecat': [],
'details': {},
'laer': [],
'energystar': {},
}
try:
params = {
"manufacturer": self.object.manufacturer,
"model": self.object.model,
}
self.params = json.dumps(params)
manuals['ifixit'] = self.request_manuals('ifixit')
manuals['icecat'] = self.request_manuals('icecat')
manuals['laer'] = self.request_manuals('laer')
manuals['energystar'] = self.request_manuals('energystar') or {}
if manuals['icecat']:
manuals['details'] = manuals['icecat'][0]
except Exception as err:
logger.error("Error: {}".format(err))
self.context['manuals'] = manuals
self.parse_energystar()
def parse_energystar(self):
if not self.context.get('manuals', {}).get('energystar'):
return
# Defined in:
# https://dev.socrata.com/foundry/data.energystar.gov/j7nq-iepp
energy_types = [
'functional_adder_allowances_kwh',
'tec_allowance_kwh',
'long_idle_watts',
'short_idle_watts',
'off_mode_watts',
'sleep_mode_watts',
'tec_of_model_kwh',
'tec_requirement_kwh',
'work_off_mode_watts',
'work_weighted_power_of_model_watts',
]
energy = {}
for field in energy_types:
energy[field] = []
for e in self.context['manuals']['energystar']:
for field in energy_types:
for k, v in e.items():
if not v:
continue
if field in k:
energy[field].append(v)
for k, v in energy.items():
if not v:
energy[k] = 0
continue
tt = sum([float(i) for i in v])
energy[k] = round(tt / len(v), 2)
self.context['manuals']['energystar'] = energy
def request_manuals(self, prefix):
#TODO reimplement manuals service
response = {
"laer": [{"metal": 40, "plastic_post_consumer": 27, "plastic_post_industry": 34}],
"energystar": [{
'functional_adder_allowances_kwh': 180,
"long_idle_watts": 240,
"short_idle_watts": 120,
"sleep_mode_watts": 30,
"off_mode_watts": 3,
"tec_allowance_kwh": 180,
"tec_of_model_kwh": 150,
"tec_requirement_kwh": 220,
"work_off_mode_watts": 70,
"work_weighted_power_of_model_watts": 240
}],
"ifixit": [
{
"image": "https://guide-images.cdn.ifixit.com/igi/156EpI4YdQeVfVPa.medium",
"url": "https://es.ifixit.com/Gu%C3%ADa/HP+ProBook+450+G4+Back+Panel+Replacement/171196?lang=en",
"title": "HP ProBook 450 G4 Back Panel Replacement"
},
{
"image": "https://guide-images.cdn.ifixit.com/igi/usTIqCKpuxVWC3Ix.140x105",
"url": "https://es.ifixit.com/Gu%C3%ADa/HP+ProBook+450+G4+Display+Assembly+Replacement/171101?lang=en",
"title": "Display Assembly Replacement"
}
],
"icecat": [
{
"logo": "https://images.icecat.biz/img/brand/thumb/1_cf8603f6de7b4c4d8ac4f5f0ef439a05.jpg",
"image": "https://guide-images.cdn.ifixit.com/igi/Q2nYjTIQfG6GaI5B.standard",
"pdf": "https://icecat.biz/rest/product-pdf?productId=32951710&lang=en",
"title": "HP ProBook 450 G3"
}
]
}
return response.get(prefix, {})

View file

@ -9,6 +9,7 @@ services:
- ALLOWED_HOSTS=${ALLOWED_HOSTS:-$DOMAIN}
- DEMO=${DEMO:-false}
- PREDEFINED_TOKEN=${PREDEFINED_TOKEN:-}
- DPP=${DPP:-false}
volumes:
- .:/opt/devicehub-django
ports:

View file

@ -20,7 +20,7 @@ main() {
echo "WARNING: .env was not there, .env.example was copied, this only happens once"
fi
# remove old database
sudo rm -vfr ./db/*
rm -vfr ./db/*
docker compose down -v
docker compose build
docker compose up ${detach_arg:-}

View file

@ -7,8 +7,14 @@ RUN apt update && \
git \
sqlite3 \
jq \
time \
vim \
&& rm -rf /var/lib/apt/lists/*
# TODO I don't like this, but the whole ereuse-dpp works with user 1000 because of the volume mapping
# thanks https://stackoverflow.com/questions/70520205/docker-non-root-user-best-practices-for-python-images
RUN adduser --home /opt/devicehub-django -u 1000 app
WORKDIR /opt/devicehub-django
# reduce size (python specifics) -> src https://stackoverflow.com/questions/74616667/removing-pip-cache-after-installing-dependencies-in-docker-image
@ -27,10 +33,16 @@ RUN python -m pip install --upgrade pip || (rm -rf /usr/local/lib/python3.11/sit
COPY ./requirements.txt /opt/devicehub-django
RUN pip install -r requirements.txt
# TODO hardcoded, is ignored in requirements.txt
RUN pip install -i https://test.pypi.org/simple/ ereuseapitest==0.0.14
# TODO Is there a better way?
# Set PYTHONPATH to include the directory with the xapian module
ENV PYTHONPATH="${PYTHONPATH}:/usr/lib/python3/dist-packages"
COPY docker/devicehub-django.entrypoint.sh /
RUN chown -R app:app /opt/devicehub-django
USER app
ENTRYPOINT sh /devicehub-django.entrypoint.sh

View file

@ -5,6 +5,150 @@ set -u
# DEBUG
set -x
# TODO there is a conflict between two shared vars
# 1. from the original docker compose devicehub-teal
# 2. from the new docker compose that integrates all dpp services
wait_for_dpp_shared() {
while true; do
# specially ensure VERAMO_API_CRED_FILE is not empty,
# it takes some time to get data in
OPERATOR_TOKEN_FILE='operator-token.txt'
if [ -f "/shared/${OPERATOR_TOKEN_FILE}" ] && \
[ -f "/shared/create_user_operator_finished" ]; then
sleep 5
echo "Files ready to process."
break
else
echo "Waiting for file in shared: ${OPERATOR_TOKEN_FILE}"
sleep 5
fi
done
}
# 3. Generate an environment .env file.
# TODO cargar via shared
gen_env_vars() {
INIT_ORG="${INIT_ORG:-example-org}"
INIT_USER="${INIT_USER:-user@example.org}"
INIT_PASSWD="${INIT_PASSWD:-1234}"
ADMIN='True'
PREDEFINED_TOKEN="${PREDEFINED_TOKEN:-}"
# specific dpp env vars
if [ "${DPP:-}" = 'true' ]; then
# fill env vars in this docker entrypoint
wait_for_dpp_shared
export API_DLT='http://api_connector:3010'
export API_DLT_TOKEN="$(cat "/shared/${OPERATOR_TOKEN_FILE}")"
export API_RESOLVER='http://id_index_api:3012'
# TODO hardcoded
export ID_FEDERATED='DH1'
# propagate to .env
dpp_env_vars="$(cat <<END
API_DLT=${API_DLT}
API_DLT_TOKEN=${API_DLT_TOKEN}
API_RESOLVER=${API_RESOLVER}
ID_FEDERATED=${ID_FEDERATED}
END
)"
fi
# generate config using env vars from docker
# TODO rethink if this is needed because now this is django, not flask
cat > .env <<END
${dpp_env_vars:-}
END
}
handle_federated_id() {
# devicehub host and id federated checker
# //getAll queries are not accepted by this service, so we remove them
EXPECTED_ID_FEDERATED="$(curl -s "${API_RESOLVER%/}/getAll" \
| jq -r '.url | to_entries | .[] | select(.value == "'"${DEVICEHUB_HOST}"'") | .key' \
| head -n 1)"
# if is a new DEVICEHUB_HOST, then register it
if [ -z "${EXPECTED_ID_FEDERATED}" ]; then
# TODO better docker compose run command
cmd="docker compose run --entrypoint= devicehub flask dlt_insert_members ${DEVICEHUB_HOST}"
big_error "No FEDERATED ID maybe you should run \`${cmd}\`"
fi
# if not new DEVICEHUB_HOST, then check consistency
# if there is already an ID in the DLT, it should match with my internal ID
if [ ! "${EXPECTED_ID_FEDERATED}" = "${ID_FEDERATED}" ]; then
big_error "ID_FEDERATED should be ${EXPECTED_ID_FEDERATED} instead of ${ID_FEDERATED}"
fi
# not needed, but reserved
# EXPECTED_DEVICEHUB_HOST="$(curl -s "${API_RESOLVER%/}/getAll" \
# | jq -r '.url | to_entries | .[] | select(.key == "'"${ID_FEDERATED}"'") | .value' \
# | head -n 1)"
# if [ ! "${EXPECTED_DEVICEHUB_HOST}" = "${DEVICEHUB_HOST}" ]; then
# big_error "ERROR: DEVICEHUB_HOST should be ${EXPECTED_DEVICEHUB_HOST} instead of ${DEVICEHUB_HOST}"
# fi
}
config_dpp_part1() {
# 12. Add a new server to the 'api resolver'
if [ "${ID_SERVICE:-}" ]; then
handle_federated_id
else
# TODO when this runs more than one time per service, this is a problem, but for the docker-reset.sh workflow, that's fine
# TODO put this in already_configured
# TODO hardcoded http proto and port
./manage.py dlt_insert_members "http://${DOMAIN}:8000"
fi
# 13. Do a rsync api resolve
./manage.py dlt_rsync_members
# 14. Register a new user to the DLT
DATASET_FILE='/tmp/dataset.json'
cat > "${DATASET_FILE}" <<END
{
"email": "${INIT_USER}",
"password": "${INIT_PASSWD}",
"api_token": "${API_DLT_TOKEN}"
}
END
./manage.py dlt_register_user "${DATASET_FILE}"
}
config_phase() {
# TODO review this flag file
init_flagfile="${program_dir}/already_configured"
if [ ! -f "${init_flagfile}" ]; then
# non DL user (only for the inventory)
./manage.py add_institution "${INIT_ORG}"
# TODO: one error on add_user, and you don't add user anymore
./manage.py add_user "${INIT_ORG}" "${INIT_USER}" "${INIT_PASSWD}" "${ADMIN}" "${PREDEFINED_TOKEN}"
if [ "${DPP:-}" = 'true' ]; then
# 12, 13, 14
config_dpp_part1
# cleanup other spnapshots and copy dlt/dpp snapshots
# TODO make this better
rm example/snapshots/*
cp example/dpp-snapshots/*.json example/snapshots/
fi
# # 15. Add inventory snapshots for user "${INIT_USER}".
if [ "${DEMO:-}" = 'true' ]; then
/usr/bin/time ./manage.py up_snapshots example/snapshots/ "${INIT_USER}"
fi
# remain next command as the last operation for this if conditional
touch "${init_flagfile}"
fi
}
check_app_is_there() {
if [ ! -f "./manage.py" ]; then
usage
@ -13,7 +157,7 @@ check_app_is_there() {
deploy() {
# TODO this is weird, find better workaround
git config --global --add safe.directory /opt/devicehub-django
git config --global --add safe.directory "${program_dir}"
export COMMIT=$(git log --format="%H %ad" --date=iso -n 1)
if [ "${DEBUG:-}" = 'true' ]; then
@ -31,18 +175,7 @@ deploy() {
# inspired by https://medium.com/analytics-vidhya/django-with-docker-and-docker-compose-python-part-2-8415976470cc
echo "INFO detected NEW deployment"
./manage.py migrate
INIT_ORG="${INIT_ORG:-example-org}"
INIT_USER="${INIT_USER:-user@example.org}"
INIT_PASSWD="${INIT_PASSWD:-1234}"
ADMIN='True'
PREDEFINED_TOKEN="${PREDEFINED_TOKEN:-}"
./manage.py add_institution "${INIT_ORG}"
# TODO: one error on add_user, and you don't add user anymore
./manage.py add_user "${INIT_ORG}" "${INIT_USER}" "${INIT_PASSWD}" "${ADMIN}" "${PREDEFINED_TOKEN}"
if [ "${DEMO:-}" = 'true' ]; then
./manage.py up_snapshots example/snapshots/ "${INIT_USER}"
fi
config_phase
fi
}
@ -70,6 +203,7 @@ runserver() {
main() {
program_dir='/opt/devicehub-django'
cd "${program_dir}"
gen_env_vars
deploy
runserver
}

0
dpp/__init__.py Normal file
View file

3
dpp/admin.py Normal file
View file

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

166
dpp/api_dlt.py Normal file
View file

@ -0,0 +1,166 @@
import json
import time
import logging
from django.conf import settings
from ereuseapi.methods import API
from dpp.models import Proof, UserDpp
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(user):
dp = UserDpp.objects.filter(user=user).first()
if not dp:
return
api_dlt = settings.API_DLT
token_dlt = dp.api_keys_dlt
if not api_dlt or not token_dlt:
logger.error("NOT POSSIBLE CONNECT WITH API DLT!!!")
return
return API(api_dlt, token_dlt, "ethereum")
def register_dlt(api, chid, phid, proof_type=None):
if proof_type:
return api.generate_proof(
chid,
ALGORITHM,
phid,
proof_type,
settings.ID_FEDERATED
)
return api.register_device(
chid,
ALGORITHM,
phid,
settings.ID_FEDERATED
)
def issuer_dpp_dlt(api, dpp):
phid = dpp.split(":")[1]
return api.issue_passport(
dpp,
ALGORITHM,
phid,
settings.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
logger.debug("timestamp: %s", timestamp)
d = {
"type": proof_type,
"timestamp": timestamp,
"issuer": user.institution,
"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 = connect_api(user)
if not api:
cny_a = 0
return
result = register_dlt(api, 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.ID_FEDERATED:
cny = 1
while cny:
try:
api.add_service(
chid,
'DeviceHub',
settings.ID_FEDERATED,
'Inventory service',
'Inv',
)
cny = 0
except Exception:
time.sleep(10)
def register_passport_dlt(chid, phid, ev_uuid, user):
token_dlt = settings.TOKEN_DLT
api_dlt = settings.API_DLT
if not token_dlt or not api_dlt:
return
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:
api = connect_api(user)
if not api:
cny_a = 0
return
result = issuer_dpp_dlt(api, 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)

6
dpp/apps.py Normal file
View file

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

View file

@ -0,0 +1,35 @@
import logging
import requests
from django.core.management.base import BaseCommand
from django.conf import settings
from user.models import Institution
logger = logging.getLogger('django')
class Command(BaseCommand):
help = "Insert a new Institution in DLT"
def add_arguments(self, parser):
parser.add_argument('domain', type=str, help='institution')
def handle(self, *args, **kwargs):
domain = kwargs.get("domain")
api = settings.API_RESOLVER
if not api:
logger.error("you need set the var API_RESOLVER")
return
if "http" not in domain:
logger.error("you need put https:// in %s", domain)
return
api = api.strip("/")
domain = domain.strip("/")
data = {"url": domain}
url = api + '/registerURL'
res = requests.post(url, json=data)
print(res.json())

View file

@ -0,0 +1,72 @@
import json
import logging
from ereuseapi.methods import API
from django.conf import settings
from django.core.management.base import BaseCommand
from user.models import User, Institution
from dpp.models import UserDpp
logger = logging.getLogger('django')
class Command(BaseCommand):
help = "Insert users than are in Dlt with params: path of data set file"
def add_arguments(self, parser):
parser.add_argument('dataset_file', type=str, help='institution')
def handle(self, *args, **kwargs):
dataset_file = kwargs.get("dataset_file")
self.api_dlt = settings.API_DLT
self.institution = Institution.objects.filter().first()
if not self.api_dlt:
logger.error("you need set the var API_DLT")
return
self.api_dlt = self.api_dlt.strip("/")
with open(dataset_file) as f:
dataset = json.loads(f.read())
self.add_user(dataset)
def add_user(self, data):
email = data.get("email")
password = data.get("password")
api_token = data.get("api_token")
# ethereum = {"data": {"api_token": api_token}}
# data_eth = json.dumps(ethereum)
data_eth = json.dumps(api_token)
# TODO encrypt in the future
# api_keys_dlt = encrypt(password, data_eth)
api_keys_dlt = data_eth.strip('"').strip("'")
user = User.objects.filter(email=email).first()
if not user:
user = User.objects.create(
email=email,
password=password,
institution = self.institution
)
roles = []
token_dlt = api_token
api = API(self.api_dlt, token_dlt, "ethereum")
result = api.check_user_roles()
if result.get('Status') == 200:
if 'Success' in result.get('Data', {}).get('status'):
rols = result.get('Data', {}).get('data', {})
roles = [(k, k) for k, v in rols.items() if v]
roles_dlt = json.dumps(roles)
UserDpp.objects.create(
roles_dlt=roles_dlt,
api_keys_dlt=api_keys_dlt,
user=user
)

View file

@ -0,0 +1,47 @@
import logging
import requests
from django.core.management.base import BaseCommand
from django.conf import settings
from dpp.models import MemberFederated
logger = logging.getLogger('django')
class Command(BaseCommand):
help = "Synchronize members of DLT"
def handle(self, *args, **kwargs):
api = settings.API_RESOLVER
if not api:
logger.error("you need set the var API_RESOLVER")
return
api = api.strip("/")
url = api + '/getAll'
res = requests.get(url)
if res.status_code != 200:
return "Error, {}".format(res.text)
response = res.json()
members = response['url']
counter = members.pop('counter')
if counter <= MemberFederated.objects.count():
logger.info("Synchronize members of DLT -> All Ok")
return "All ok"
for k, v in members.items():
id = self.clean_id(k)
member = MemberFederated.objects.filter(dlt_id_provider=id).first()
if member:
if member.domain != v:
member.domain = v
member.save()
continue
MemberFederated.objects.create(dlt_id_provider=id, domain=v)
return res.text
def clean_id(self, id):
return int(id.split('DH')[-1])

View file

@ -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,
),
),
],
),
]

View file

@ -0,0 +1,25 @@
# Generated by Django 5.0.6 on 2024-11-19 19:18
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
("dpp", "0001_initial"),
]
operations = [
migrations.CreateModel(
name="MemberFederated",
fields=[
(
"dlt_id_provider",
models.IntegerField(primary_key=True, serialize=False),
),
("domain", models.CharField(max_length=256)),
("client_id", models.CharField(max_length=256)),
("client_secret", models.CharField(max_length=256)),
],
),
]

View file

@ -0,0 +1,60 @@
# Generated by Django 5.0.6 on 2024-11-20 10:51
import django.db.models.deletion
from django.conf import settings
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
("dpp", "0002_memberfederated"),
("user", "0001_initial"),
migrations.swappable_dependency(settings.AUTH_USER_MODEL),
]
operations = [
migrations.AddField(
model_name="memberfederated",
name="institution",
field=models.ForeignKey(
blank=True,
null=True,
on_delete=django.db.models.deletion.SET_NULL,
to="user.institution",
),
),
migrations.AlterField(
model_name="memberfederated",
name="client_id",
field=models.CharField(max_length=256, null=True),
),
migrations.AlterField(
model_name="memberfederated",
name="client_secret",
field=models.CharField(max_length=256, null=True),
),
migrations.CreateModel(
name="UserDpp",
fields=[
(
"id",
models.BigAutoField(
auto_created=True,
primary_key=True,
serialize=False,
verbose_name="ID",
),
),
("roles_dlt", models.TextField()),
("api_keys_dlt", models.TextField()),
(
"user",
models.ForeignKey(
on_delete=django.db.models.deletion.CASCADE,
to=settings.AUTH_USER_MODEL,
),
),
],
),
]

View file

32
dpp/models.py Normal file
View file

@ -0,0 +1,32 @@
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)
class MemberFederated(models.Model):
dlt_id_provider = models.IntegerField(primary_key=True)
domain = models.CharField(max_length=STR_EXTEND_SIZE)
# This client_id and client_secret is used for connected to this domain as
# a client and this domain then is the server of auth
client_id = models.CharField(max_length=STR_EXTEND_SIZE, null=True)
client_secret = models.CharField(max_length=STR_EXTEND_SIZE, null=True)
institution = models.ForeignKey(
Institution, on_delete=models.SET_NULL, null=True, blank=True)
class UserDpp(models.Model):
roles_dlt = models.TextField()
api_keys_dlt = models.TextField()
user = models.ForeignKey(User, on_delete=models.CASCADE)

3
dpp/tests.py Normal file
View file

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

8
dpp/urls.py Normal file
View file

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

40
dpp/views.py Normal file
View file

@ -0,0 +1,40 @@
import json
import logging
import hashlib
from django.views.generic.edit import View
from django.http import JsonResponse
from dpp.api_dlt import ALGORITHM
from evidence.models import Evidence
from evidence.parse import Build
from dpp.models import Proof
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 = Evidence(proof.uuid)
if not ev.doc:
return JsonResponse({}, status=404)
dev = Build(ev.doc, None, check=True)
doc = dev.get_phid()
data = {
"algorithm": ALGORITHM,
"document": json.dumps(doc)
}
d = {
'@context': ['https://ereuse.org/proof0.json'],
'data': data,
}
response = JsonResponse(d, status=200)
response["Access-Control-Allow-Origin"] = "*"
return response

View file

@ -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:

View file

@ -4,15 +4,19 @@ import logging
from dmidecode import DMIParse
from json_repair import repair_json
from evidence.parse_details import get_lshw_child
from django.conf import settings
from evidence.parse_details import get_lshw_child, ParseSnapshot
from evidence.models import Annotation
from evidence.xapian import index
from utils.constants import CHASSIS_DH
if settings.DPP:
from dpp.api_dlt import register_device_dlt, register_passport_dlt
logger = logging.getLogger('django')
def get_mac(lshw):
try:
if type(lshw) is dict:
@ -40,6 +44,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 +53,8 @@ class Build:
self.index()
self.create_annotations()
if settings.DPP:
self.register_device_dlt()
def index(self):
snap = json.dumps(self.json)
@ -55,6 +63,7 @@ class Build:
def generate_chids(self):
self.algorithms = {
'hidalgo1': self.get_hid_14(),
'legacy_dpp': self.get_chid_dpp(),
}
def get_hid_14(self):
@ -70,7 +79,61 @@ 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 get_chid_dpp(self):
if self.json.get("software") == "workbench-script":
dmidecode_raw = self.json["data"]["dmidecode"]
dmi = DMIParse(dmidecode_raw)
manufacturer = dmi.manufacturer().strip()
model = dmi.model().strip()
chassis = self.get_chassis_dh()
serial_number = dmi.serial_number()
sku = self.get_sku()
typ = chassis
version = self.get_version()
hid = f"{manufacturer}{model}{chassis}{serial_number}{sku}{typ}{version}"
else:
device = self.json['device']
hid = self.get_id_hw_dpp(device)
self.chid = hashlib.sha3_256(hid.encode("utf-8")).hexdigest()
return self.chid
def get_id_hw_dpp(self, d):
manufacturer = d.get("manufacturer", '')
model = d.get("model", '')
chassis = d.get("chassis", '')
serial_number = d.get("serialNumber", '')
sku = d.get("sku", '')
typ = d.get("type", '')
version = d.get("version", '')
return f"{manufacturer}{model}{chassis}{serial_number}{sku}{typ}{version}"
def get_phid(self):
if self.json.get("software") == "workbench-script":
data = ParseSnapshot(self.json)
self.device = data.device
self.components = data.components
else:
self.device = self.json.get("device")
self.components = self.json.get("components", [])
self.device.pop("actions", None)
for c in self.components:
c.pop("actions", None)
device = self.get_id_hw_dpp(self.device)
components = sorted(self.components, key=lambda x: x.get("type"))
doc = [("computer", device)]
for c in components:
doc.append((c.get("type"), self.get_id_hw_dpp(c)))
return doc
def create_annotations(self):
annotation = Annotation.objects.filter(
@ -106,7 +169,10 @@ class Build:
return self.dmi.get("System")[0].get("SKU Number", "n/a").strip()
def get_chassis(self):
return self.dmi.get("Chassis")[0].get("Type", '_virtual')
return self.dmi.get("Chassis")[0].get("Type", '_virtual') #
def get_version(self):
return self.dmi.get("System")[0].get("Verson", '_virtual')
def get_hid(self, snapshot):
dmidecode_raw = snapshot["data"]["dmidecode"]
@ -129,3 +195,12 @@ 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):
chid = self.algorithms.get('legacy_dpp')
phid = self.get_signature(self.get_phid())
register_device_dlt(chid, phid, self.uuid, self.user)
register_passport_dlt(chid, phid, self.uuid, self.user)

View file

@ -137,7 +137,7 @@ class DownloadEvidenceView(DashboardView, TemplateView):
evidence.get_doc()
data = json.dumps(evidence.doc)
response = HttpResponse(data, content_type="application/json")
response['Content-Disposition'] = 'attachment; filename={}'.format("credential.json")
response['Content-Disposition'] = 'attachment; filename={}'.format("evidence.json")
return response

View file

@ -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)
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)
enquire = xapian.Enquire(database)
enquire.set_query(final_query)
matches = enquire.get_mset(offset, limit)

File diff suppressed because one or more lines are too long

File diff suppressed because one or more lines are too long

File diff suppressed because one or more lines are too long

File diff suppressed because one or more lines are too long

View file

@ -0,0 +1 @@
{"closed": true, "components": [{"actions": [], "manufacturer": "Intel Corporation", "model": "82579LM Gigabit Network Connection", "serialNumber": "00:11:11:11:11:00", "speed": 1000.0, "type": "NetworkAdapter", "variant": "04", "wireless": false}, {"actions": [], "manufacturer": "Intel Corporation", "model": "7 Series/C216 Chipset Family High Definition Audio Controller", "serialNumber": null, "type": "SoundCard"}, {"actions": [], "format": "DIMM", "interface": "DDR3", "manufacturer": "Micron", "model": "16KTF51264AZ", "serialNumber": "AAAAAAAA", "size": 4096.0, "speed": 1600.0, "type": "RamModule"}, {"actions": [{"endTime": "2022-10-11T13:45:31.239555+00:00", "severity": "Info", "startTime": "2021-10-11T09:45:19.623967+00:00", "steps": [{"endTime": "2021-10-11T11:05:28.090897+00:00", "severity": "Info", "startTime": "2021-10-11T09:45:19.624163+00:00", "type": "StepZero"}, {"endTime": "2021-10-11T13:45:31.239402+00:00", "severity": "Info", "startTime": "2021-10-11T11:05:28.091255+00:00", "type": "StepRandom"}], "type": "EraseSectors"}, {"assessment": true, "commandTimeout": 30, "currentPendingSectorCount": 0, "elapsed": 60, "length": "Short", "lifetime": 18720, "offlineUncorrectable": 0, "powerCycleCount": 2147, "reallocatedSectorCount": 0, "reportedUncorrectableErrors": 0, "severity": "Info", "status": "Completed without error", "type": "TestDataStorage"}, {"elapsed": 11, "readSpeed": 119.0, "type": "BenchmarkDataStorage", "writeSpeed": 32.7}], "interface": "ATA", "manufacturer": "Seagate", "model": "ST3500418AS", "serialNumber": "AAAAAAAA", "size": 500000.0, "type": "HardDrive", "variant": "CC46"}, {"actions": [{"elapsed": 0, "rate": 25540.36, "type": "BenchmarkProcessor"}, {"elapsed": 8, "rate": 7.6939, "type": "BenchmarkProcessorSysbench"}], "address": 64, "brand": "Core i5", "cores": 4, "generation": 3, "manufacturer": "Intel Corp.", "model": "Intel Core i5-3470 CPU @ 3.20GHz", "serialNumber": null, "speed": 1.6242180000000002, "threads": 4, "type": "Processor"}, {"actions": [], "manufacturer": "Intel Corporation", "memory": null, "model": "Xeon E3-1200 v2/3rd Gen Core processor Graphics Controller", "serialNumber": null, "type": "GraphicCard"}, {"actions": [], "biosDate": "2012-08-07T00:00:00", "firewire": 0, "manufacturer": "LENOVO", "model": "MAHOBAY", "pcmcia": 0, "ramMaxSize": 32, "ramSlots": 4, "serial": 1, "serialNumber": null, "slots": 4, "type": "Motherboard", "usb": 3, "version": "9SKT39AUS"}], "device": {"actions": [{"elapsed": 1, "rate": 0.6507, "type": "BenchmarkRamSysbench"}], "chassis": "Tower", "manufacturer": "LENOVO", "model": "3227A2G", "serialNumber": "AAAAAAAA", "sku": "LENOVO_MT_3227", "type": "Desktop", "version": "ThinkCentre M92P"}, "elapsed": 187302510, "endTime": "2016-11-03T17:17:01.116554+00:00", "software": "Workbench", "type": "Snapshot", "uuid": "ae913de1-e639-476a-ad9b-78eabbe4628b", "version": "11.0b11"}

View file

@ -0,0 +1 @@
{"closed": true, "components": [{"actions": [], "manufacturer": "Intel Corporation", "model": "82579LM Gigabit Network Connection", "serialNumber": "00:11:11:11:11:00", "speed": 1000.0, "type": "NetworkAdapter", "variant": "04", "wireless": false}, {"actions": [], "manufacturer": "Intel Corporation", "model": "7 Series/C216 Chipset Family High Definition Audio Controller", "serialNumber": null, "type": "SoundCard"}, {"actions": [], "format": "DIMM", "interface": "DDR3", "manufacturer": "Micron", "model": "16KTF51264AZ", "serialNumber": "AAAAAAAA", "size": 4096.0, "speed": 1600.0, "type": "RamModule"}, {"actions": [{"endTime": "2022-10-11T13:45:31.239555+00:00", "severity": "Info", "startTime": "2021-10-11T09:45:19.623967+00:00", "steps": [{"endTime": "2021-10-11T11:05:28.090897+00:00", "severity": "Info", "startTime": "2021-10-11T09:45:19.624163+00:00", "type": "StepZero"}, {"endTime": "2021-10-11T13:45:31.239402+00:00", "severity": "Info", "startTime": "2021-10-11T11:05:28.091255+00:00", "type": "StepRandom"}], "type": "EraseSectors"}, {"assessment": true, "commandTimeout": 30, "currentPendingSectorCount": 0, "elapsed": 60, "length": "Short", "lifetime": 18720, "offlineUncorrectable": 0, "powerCycleCount": 2147, "reallocatedSectorCount": 0, "reportedUncorrectableErrors": 0, "severity": "Info", "status": "Completed without error", "type": "TestDataStorage"}, {"elapsed": 11, "readSpeed": 119.0, "type": "BenchmarkDataStorage", "writeSpeed": 32.7}], "interface": "ATA", "manufacturer": "Seagate", "model": "ST3500418AS", "serialNumber": "AAAAAAAA", "size": 500000.0, "type": "HardDrive", "variant": "CC46"}, {"actions": [{"elapsed": 0, "rate": 25540.36, "type": "BenchmarkProcessor"}, {"elapsed": 8, "rate": 7.6939, "type": "BenchmarkProcessorSysbench"}], "address": 64, "brand": "Core i5", "cores": 4, "generation": 3, "manufacturer": "Intel Corp.", "model": "Intel Core i5-3470 CPU @ 3.20GHz", "serialNumber": null, "speed": 1.6242180000000002, "threads": 4, "type": "Processor"}, {"actions": [], "manufacturer": "Intel Corporation", "memory": null, "model": "Xeon E3-1200 v2/3rd Gen Core processor Graphics Controller", "serialNumber": null, "type": "GraphicCard"}, {"actions": [], "biosDate": "2012-08-07T00:00:00", "firewire": 0, "manufacturer": "LENOVO", "model": "MAHOBAY", "pcmcia": 0, "ramMaxSize": 32, "ramSlots": 4, "serial": 1, "serialNumber": null, "slots": 4, "type": "Motherboard", "usb": 3, "version": "9SKT39AUS"}], "device": {"actions": [{"elapsed": 1, "rate": 0.6507, "type": "BenchmarkRamSysbench"}], "chassis": "Tower", "manufacturer": "LENOVO", "model": "3227A2G", "serialNumber": "AAAAAAAAD", "sku": "LENOVO_MT_3227", "type": "Desktop", "version": "ThinkCentre M92P"}, "elapsed": 187302510, "endTime": "2016-11-03T17:17:01.116554+00:00", "software": "Workbench", "type": "Snapshot", "uuid": "ae913de1-e639-476a-ad9b-78eabbe4625b", "version": "11.0b11"}

View file

@ -11,3 +11,6 @@ xlrd==2.0.1
odfpy==1.4.1
pytz==2024.2
json-repair==0.30.0
setuptools==75.5.0
requests==2.32.3
wheel==0.45.0

View file

@ -17,8 +17,19 @@ HID_ALGO1 = [
"sku"
]
LEGACY_DPP = [
"manufacturer",
"model",
"chassis",
"serialNumber",
"sku",
"type",
"version"
]
ALGOS = {
"hidalgo1": HID_ALGO1,
"legacy_dpp": LEGACY_DPP
}

View file

@ -22,16 +22,24 @@ class CustomFormatter(logging.Formatter):
record.levelname = f"{color}{record.levelname}{RESET}"
if record.args:
record.msg = self.highlight_args(record.msg, record.args, color)
try:
record.msg = record.msg % record.args
record.args = ()
except (TypeError, ValueError):
record.msg = f"{color}{record.msg}{RESET}"
# provide trace when DEBUG config
if settings.DEBUG:
import traceback
print(traceback.format_exc())
# Highlight the final formatted message
record.msg = self.highlight_message(record.msg, color)
# pedro says: I discovered that trace is provided anyway with
# this commented (reason: strange None msgs)
# is this needed?
### provide trace when DEBUG config
#if settings.DEBUG:
# import traceback
# print(traceback.format_exc())
return super().format(record)
def highlight_args(self, message, args, color):
highlighted_args = tuple(f"{color}{arg}{RESET}" for arg in args)
return message % highlighted_args
def highlight_message(self, message, color):
return f"{color}{message}{RESET}"