split parse files for cases of use

evidence/forms.py

@@ -8,7 +8,6 @@ from utils.device import create_annotation, create_doc, create_index
 from utils.forms import MultipleFileField
 from device.models import Device
 from evidence.parse import Build
-from evidence.legacy_parse import Build as legacy_build
 from evidence.models import Annotation
 from utils.save_snapshots import move_json, save_in_disk
 
@@ -31,10 +30,6 @@ class UploadForm(forms.Form):
             try:
                 file_json = json.loads(file_data)
                 build = Build
-                if file_json.get("data",{}).get("lshw"):
-                    if file_json.get("software") == "workbench-script":
-                        build  = legacy_build
-
                 snap = build(file_json, None, check=True)
                 exist_annotation = Annotation.objects.filter(
                     uuid=snap.uuid
@@ -65,10 +60,6 @@ class UploadForm(forms.Form):
             path_name = save_in_disk(ev[1], user.institution.name)
             build = Build
             file_json = ev[1]
-            if file_json.get("data",{}).get("lshw"):
-                if file_json.get("software") == "workbench-script":
-                    build  = legacy_build
-
             build(file_json, user)
             move_json(path_name, user.institution.name)
 

evidence/legacy_parse.py

@@ -1,17 +1,11 @@
 import json
-import hashlib
 import logging
 
 from dmidecode import DMIParse
 from json_repair import repair_json
-from django.conf import settings
+from evidence.mixin_parse import BuildMix
 from evidence.legacy_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')
@@ -37,100 +31,21 @@ def get_mac(lshw):
         return mac
 
 
-class Build:
+class Build(BuildMix):
-    def __init__(self, evidence_json, user, check=False):
+    # This parse is for get info from snapshots created with
-        self.json = evidence_json
+    # workbench-script but builded for send to devicehub-teal
-        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:
+    def get_details(self):
-            return
-
-        self.index()
-        self.create_annotations()
-        if settings.DPP:
-            self.register_device_dlt()
-
-    def index(self):
-        snap = json.dumps(self.json)
-        index(self.user.institution, self.uuid, snap)
-
-    def generate_chids(self):
-        self.algorithms = {
-            'hidalgo1': self.get_hid(),
-            'legacy_dpp': self.get_chid_dpp(),
-        }
-
-    def get_chid_dpp(self):
         dmidecode_raw = self.json["data"]["dmidecode"]
-        dmi = DMIParse(dmidecode_raw)
+        self.dmi = DMIParse(dmidecode_raw)
 
-        manufacturer = dmi.manufacturer().strip()
+        self.manufacturer = self.dmi.manufacturer().strip()
-        model = dmi.model().strip()
+        self.model = self.dmi.model().strip()
-        chassis = self.get_chassis_dh()
+        self.chassis = self.get_chassis_dh()
-        serial_number = dmi.serial_number()
+        self.serial_number = self.dmi.serial_number()
-        sku = self.get_sku()
+        self.sku = self.get_sku()
-        typ = chassis
+        self.typ = self.chassis
-        version = self.get_version()
+        self.version = self.get_version()
-        hid = f"{manufacturer}{model}{chassis}{serial_number}{sku}{typ}{version}"
-
-        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):
-        data = ParseSnapshot(self.json)
-        self.device = data.device
-        self.components = data.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(
-                uuid=self.uuid,
-                owner=self.user.institution,
-                type=Annotation.Type.SYSTEM,
-        )
-
-        if annotation:
-            txt = "Warning: Snapshot %s already registered (annotation exists)"
-            logger.warning(txt, self.uuid)
-            return
-
-        for k, v in self.algorithms.items():
-            Annotation.objects.create(
-                uuid=self.uuid,
-                owner=self.user.institution,
-                user=self.user,
-                type=Annotation.Type.SYSTEM,
-                key=k,
-                value=v
-            )
 
     def get_chassis_dh(self):
         chassis = self.get_chassis()
@@ -149,36 +64,6 @@ class Build:
     def get_version(self):
         return self.dmi.get("System")[0].get("Verson", '_virtual')
 
-    def get_hid(self):
+    def _get_components(self):
-        snapshot = self.json
+        data = ParseSnapshot(self.json)
-        dmidecode_raw = snapshot["data"]["dmidecode"]
+        self.components = data.components
-        self.dmi = DMIParse(dmidecode_raw)
-
-        manufacturer = self.dmi.manufacturer().strip()
-        model = self.dmi.model().strip()
-        chassis = self.get_chassis_dh()
-        serial_number = self.dmi.serial_number()
-        sku = self.get_sku()
-
-        if not snapshot["data"].get('lshw'):
-            return f"{manufacturer}{model}{chassis}{serial_number}{sku}"
-
-        lshw = snapshot["data"]["lshw"]
-        # mac = get_mac2(hwinfo_raw) or ""
-        mac = get_mac(lshw) or ""
-        if not mac:
-            txt = "Could not retrieve MAC address in snapshot %s"
-            logger.warning(txt, snapshot['uuid'])
-
-        hid = f"{manufacturer}{model}{chassis}{serial_number}{sku}{mac}"
-        self.chid = hashlib.sha3_256(hid.encode()).hexdigest()
-        return self.chid
-
-    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)

evidence/legacy_parse_details.py

@@ -13,10 +13,13 @@ logger = logging.getLogger('django')
 
 
 def get_lshw_child(child, nets, component):
+    try:
         if child.get('id') == component:
             nets.append(child)
         if child.get('children'):
             [get_lshw_child(x, nets, component) for x in child['children']]
+    except Exception:
+        return []
 
 class ParseSnapshot:
     def __init__(self, snapshot, default="n/a"):
@@ -314,10 +317,14 @@ class ParseSnapshot:
 
     def get_cpu_address(self, cpu):
         default = 64
+
+        try:
             for ch in self.lshw.get('children', []):
                 for c in ch.get('children', []):
                     if c['class'] == 'processor':
                         return c.get('width', default)
+        except:
+            return default
         return default
 
     def get_usb_num(self):

evidence/mixin_parse.py

@@ -0,0 +1,58 @@
+import logging
+from django.conf import settings
+
+from utils.constants import ALGOS
+
+
+logger = logging.getLogger('django')
+
+
+class BuildMix:
+    def __init__(self, evidence_json):
+        self.json = evidence_json
+        self.uuid = self.json['uuid']
+        self.manufacturer = ""
+        self.model = ""
+        self.serial_number = ""
+        self.chassis = ""
+        self.sku = ""
+        self.mac = ""
+        self.tpy = ""
+        self.version = ""
+        self.get_details()
+        self.generate_chids()
+
+    def get_hid(self, algo):
+        algorithm = ALGOS.get(algo, [])
+        hid = ""
+        for f in algorithm:
+            if hasattr(self, f):
+                hid += getattr(self, f)
+        return hid
+
+    def generate_chids(self):
+        self.algorithms = {
+            'hidalgo1': self.get_hid('hidalgo1'),
+        }
+        if settings.DPP:
+            self.algorithms["legacy_dpp"] = self.get_hid("legacy_dpp")
+
+    def get_doc(self):
+        self._get_components()
+        for c in self.components:
+            c.pop("actions", None)
+
+        components = sorted(self.components, key=lambda x: x.get("type"))
+        device = self.algorithms.get('legacy_dpp')
+
+        doc = [("computer", device)]
+
+        for c in components:
+            doc.append((c.get("type"), self.get_id_hw_dpp(c)))
+
+    def get_id_hw_dpp(self, d):
+        algorithm = ALGOS.get("legacy_dpp", [])
+        hid = ""
+        for f in algorithm:
+            hid += d.get(f, '')
+        return hid

evidence/models.py

@@ -6,8 +6,8 @@ from django.db import models
 
 from utils.constants import STR_EXTEND_SIZE, CHASSIS_DH
 from evidence.xapian import search
-from evidence.parse_details import ParseSnapshot, get_inxi, get_inxi_key
+from evidence.parse_details import ParseSnapshot
-from evidence.legacy_parse_details import ParseSnapshot as legacy_ParseSnapshot
+from evidence.normal_parse_details import get_inxi, get_inxi_key
 from user.models import User, Institution
 
 
@@ -135,7 +135,7 @@ class Evidence:
             return list(self.doc.get('kv').values())[0]
 
         if self.is_legacy():
-            return self.doc['device']['manufacturer']
+            return self.doc.get('device', {}).get('manufacturer', '')
 
         if self.inxi:
             return self.device_manufacturer
@@ -150,7 +150,7 @@ class Evidence:
             return list(self.doc.get('kv').values())[1]
 
         if self.is_legacy():
-            return self.doc['device']['model']
+            return self.doc.get('device', {}).get('model', '')
 
         if self.inxi:
             return self.device_model
@@ -159,7 +159,7 @@ class Evidence:
 
     def get_chassis(self):
         if self.is_legacy():
-            return self.doc['device']['model']
+            return self.doc.get('device', {}).get('model', '')
 
         if self.inxi:
             return self.device_chassis
@@ -174,7 +174,7 @@ class Evidence:
 
     def get_serial_number(self):
         if self.is_legacy():
-            return self.doc['device']['serialNumber']
+            return self.doc.get('device', {}).get('serialNumber', '')
 
         if self.inxi:
             return self.device_serial_number
@@ -196,13 +196,7 @@ class Evidence:
         ).order_by("-created").values_list("uuid", "created").distinct()
 
     def set_components(self):
-        parse = ParseSnapshot
+        self.components = ParseSnapshot(self.doc).components
-        if self.doc.get("software") == "workbench-script":
-            if self.doc.get("data", {}).get("lshw"):
-                parse = legacy_ParseSnapshot
-
-        snapshot = parse(self.doc).snapshot_json
-        self.components = snapshot['components']
 
     def is_legacy(self):
         if self.doc.get("credentialSubject"):

evidence/normal_parse.py

@@ -0,0 +1,63 @@
+import json
+import logging
+
+from evidence.mixin_parse import BuildMix
+from evidence.normal_parse_details import get_inxi_key, get_inxi, ParseSnapshot
+
+
+logger = logging.getLogger('django')
+
+
+def get_mac(inxi):
+    nets = get_inxi_key(inxi, "Network")
+    networks = [(nets[i], nets[i + 1]) for i in range(0, len(nets) - 1, 2)]
+
+    for n, iface in networks:
+        if get_inxi(n, "port"):
+            return get_inxi(iface, 'mac')
+
+
+class Build(BuildMix):
+
+    def get_details(self):
+        self.from_credential()
+        try:
+            self.inxi = self.json["data"]["inxi"]
+            if isinstance(self.inxi, str):
+                self.inxi = json.loads(self.inxi)
+        except Exception:
+            logger.error("No inxi in snapshot %s", self.uuid)
+            return ""
+
+        machine = get_inxi_key(self.inxi, 'Machine')
+        for m in machine:
+            system = get_inxi(m, "System")
+            if system:
+                self.manufacturer = system
+                self.model = get_inxi(m, "product")
+                self.serial_number = get_inxi(m, "serial")
+                self.chassis = get_inxi(m, "Type")
+            else:
+                self.sku = get_inxi(m, "part-nu")
+
+        self.mac = get_mac(self.inxi) or ""
+        if not self.mac:
+            txt = "Could not retrieve MAC address in snapshot %s"
+            logger.warning(txt, self.uuid)
+
+    def from_credential(self):
+        if not self.json.get("credentialSubject"):
+            return
+
+        self.json.update(self.json["credentialSubject"])
+        if self.json.get("evidence"):
+            self.json["data"] = {}
+            for ev in self.json["evidence"]:
+                k = ev.get("operation")
+                if not k:
+                    continue
+                self.json["data"][k] = ev.get("output")
+
+    def _get_components(self):
+        data = ParseSnapshot(self.json)
+        self.components = data.components

evidence/normal_parse_details.py

@@ -0,0 +1,402 @@
+import json
+import logging
+
+from dmidecode import DMIParse
+
+
+logger = logging.getLogger('django')
+
+
+def get_inxi_key(inxi, component):
+    for n in inxi:
+        for k, v in n.items():
+            if component in k:
+                return v
+
+
+def get_inxi(n, name):
+    for k, v in n.items():
+        if f"#{name}" in k:
+            return v
+
+    return ""
+
+
+class ParseSnapshot:
+    def __init__(self, snapshot, default="n/a"):
+        self.default = default
+        self.dmidecode_raw = snapshot.get("data", {}).get("dmidecode", "{}")
+        self.smart_raw = snapshot.get("data", {}).get("smartctl", [])
+        self.inxi_raw = snapshot.get("data", {}).get("inxi", "") or ""
+        for ev in snapshot.get("evidence", []):
+            if "dmidecode" == ev.get("operation"):
+                self.dmidecode_raw = ev["output"]
+            if "inxi" == ev.get("operation"):
+                self.inxi_raw = ev["output"]
+            if "smartctl" == ev.get("operation"):
+                self.smart_raw = ev["output"]
+        data = snapshot
+        if snapshot.get("credentialSubject"):
+            data = snapshot["credentialSubject"]
+
+        self.device = {"actions": []}
+        self.components = []
+
+        self.dmi = DMIParse(self.dmidecode_raw)
+        self.smart = self.loads(self.smart_raw)
+        self.inxi = self.loads(self.inxi_raw)
+
+        self.set_computer()
+        self.set_components()
+        self.snapshot_json = {
+            "type": "Snapshot",
+            "device": self.device,
+            "software": data["software"],
+            "components": self.components,
+            "uuid": data['uuid'],
+            "endTime": data["timestamp"],
+            "elapsed": 1,
+        }
+
+    def set_computer(self):
+        machine = get_inxi_key(self.inxi, 'Machine') or []
+        for m in machine:
+            system = get_inxi(m, "System")
+            if system:
+                self.device['manufacturer'] = system
+                self.device['model'] = get_inxi(m, "product")
+                self.device['serialNumber'] = get_inxi(m, "serial")
+                self.device['type'] = get_inxi(m, "Type")
+                self.device['chassis'] = self.device['type']
+                self.device['version'] = get_inxi(m, "v")
+            else:
+                self.device['system_uuid'] = get_inxi(m, "uuid")
+                self.device['sku'] = get_inxi(m, "part-nu")
+
+    def set_components(self):
+        self.get_mother_board()
+        self.get_cpu()
+        self.get_ram()
+        self.get_graphic()
+        self.get_display()
+        self.get_networks()
+        self.get_sound_card()
+        self.get_data_storage()
+        self.get_battery()
+
+    def get_mother_board(self):
+        machine = get_inxi_key(self.inxi, 'Machine') or []
+        mb = {"type": "Motherboard",}
+        for m in machine:
+            bios_date = get_inxi(m, "date")
+            if not bios_date:
+                continue
+            mb["manufacturer"] = get_inxi(m, "Mobo")
+            mb["model"] = get_inxi(m, "model")
+            mb["serialNumber"] = get_inxi(m, "serial")
+            mb["version"] = get_inxi(m, "v")
+            mb["biosDate"] = bios_date
+            mb["biosVersion"] = self.get_bios_version()
+            mb["firewire"]: self.get_firmware_num()
+            mb["pcmcia"]: self.get_pcmcia_num()
+            mb["serial"]: self.get_serial_num()
+            mb["usb"]: self.get_usb_num()
+
+            self.get_ram_slots(mb)
+
+        self.components.append(mb)
+
+    def get_ram_slots(self, mb):
+        memory = get_inxi_key(self.inxi, 'Memory') or []
+        for m in memory:
+            slots = get_inxi(m, "slots")
+            if not slots:
+                continue
+            mb["slots"] = slots
+            mb["ramSlots"] = get_inxi(m, "modules")
+            mb["ramMaxSize"] = get_inxi(m, "capacity")
+
+
+    def get_cpu(self):
+        cpu = get_inxi_key(self.inxi, 'CPU') or []
+        cp = {"type": "Processor"}
+        vulnerabilities = []
+        for c in cpu:
+            base = get_inxi(c, "model")
+            if base:
+                cp["model"] = get_inxi(c, "model")
+                cp["arch"] = get_inxi(c, "arch")
+                cp["bits"] = get_inxi(c, "bits")
+                cp["gen"] = get_inxi(c, "gen")
+                cp["family"] = get_inxi(c, "family")
+                cp["date"] = get_inxi(c, "built")
+                continue
+            des = get_inxi(c, "L1")
+            if des:
+                cp["L1"] = des
+                cp["L2"] = get_inxi(c, "L2")
+                cp["L3"] = get_inxi(c, "L3")
+                cp["cpus"] = get_inxi(c, "cpus")
+                cp["cores"] = get_inxi(c, "cores")
+                cp["threads"] = get_inxi(c, "threads")
+                continue
+            bogo = get_inxi(c, "bogomips")
+            if bogo:
+                cp["bogomips"] = bogo
+                cp["base/boost"] = get_inxi(c, "base/boost")
+                cp["min/max"] = get_inxi(c, "min/max")
+                cp["ext-clock"] = get_inxi(c, "ext-clock")
+                cp["volts"] = get_inxi(c, "volts")
+                continue
+            ctype = get_inxi(c, "Type")
+            if ctype:
+                v = {"Type": ctype}
+                status = get_inxi(c, "status")
+                if status:
+                    v["status"] = status
+                mitigation = get_inxi(c, "mitigation")
+                if mitigation:
+                    v["mitigation"] = mitigation
+                vulnerabilities.append(v)
+
+        self.components.append(cp)
+
+
+    def get_ram(self):
+        memory = get_inxi_key(self.inxi, 'Memory') or []
+        mem = {"type": "RamModule"}
+
+        for m in memory:
+            base = get_inxi(m, "System RAM")
+            if base:
+                mem["size"] = get_inxi(m, "total")
+            slot = get_inxi(m, "manufacturer")
+            if slot:
+                mem["manufacturer"] = slot
+                mem["model"] = get_inxi(m, "part-no")
+                mem["serialNumber"] = get_inxi(m, "serial")
+                mem["speed"] = get_inxi(m, "speed")
+                mem["bits"] = get_inxi(m, "data")
+                mem["interface"] = get_inxi(m, "type")
+            module = get_inxi(m, "modules")
+            if module:
+                mem["modules"] = module
+
+        self.components.append(mem)
+
+    def get_graphic(self):
+        graphics = get_inxi_key(self.inxi, 'Graphics') or []
+
+        for c in graphics:
+            if not get_inxi(c, "Device") or not get_inxi(c, "vendor"):
+                continue
+
+            self.components.append(
+                {
+                    "type": "GraphicCard",
+                    "memory": self.get_memory_video(c),
+                    "manufacturer": get_inxi(c, "vendor"),
+                    "model": get_inxi(c, "Device"),
+                    "arch": get_inxi(c, "arch"),
+                    "serialNumber": get_inxi(c, "serial"),
+                    "integrated": True if get_inxi(c, "port") else False
+                }
+            )
+
+    def get_battery(self):
+        bats = get_inxi_key(self.inxi, 'Battery') or []
+        for b in bats:
+            self.components.append(
+                {
+                    "type": "Battery",
+                    "model": get_inxi(b, "model"),
+                    "serialNumber": get_inxi(b, "serial"),
+                    "condition": get_inxi(b, "condition"),
+                    "cycles": get_inxi(b, "cycles"),
+                    "volts": get_inxi(b, "volts")
+                }
+            )
+
+    def get_memory_video(self, c):
+        memory = get_inxi_key(self.inxi, 'Memory') or []
+
+        for m in memory:
+            igpu = get_inxi(m, "igpu")
+            agpu = get_inxi(m, "agpu")
+            ngpu = get_inxi(m, "ngpu")
+            gpu = get_inxi(m, "gpu")
+            if igpu or agpu or gpu or ngpu:
+                return igpu or agpu or gpu or ngpu
+
+        return self.default
+
+    def get_data_storage(self):
+        hdds= get_inxi_key(self.inxi, 'Drives') or []
+        for d in hdds:
+            usb = get_inxi(d, "type")
+            if usb == "USB":
+                continue
+
+            serial = get_inxi(d, "serial")
+            if serial:
+                hd = {
+                    "type": "Storage",
+                    "manufacturer": get_inxi(d, "vendor"),
+                    "model": get_inxi(d, "model"),
+                    "serialNumber": get_inxi(d, "serial"),
+                    "size": get_inxi(d, "size"),
+                    "speed": get_inxi(d, "speed"),
+                    "interface": get_inxi(d, "tech"),
+                    "firmware": get_inxi(d, "fw-rev")
+                }
+                rpm = get_inxi(d, "rpm")
+                if rpm:
+                    hd["rpm"] = rpm
+
+                family = get_inxi(d, "family")
+                if family:
+                    hd["family"] = family
+
+                sata = get_inxi(d, "sata")
+                if sata:
+                    hd["sata"] = sata
+
+                continue
+
+
+            cycles = get_inxi(d, "cycles")
+            if cycles:
+                hd['cycles'] = cycles
+                hd["health"] = get_inxi(d, "health")
+                hd["time of used"] = get_inxi(d, "on")
+                hd["read used"] = get_inxi(d, "read-units")
+                hd["written used"] = get_inxi(d, "written-units")
+
+                self.components.append(hd)
+                continue
+
+            hd = {}
+
+    def sanitize(self, action):
+        return []
+
+    def get_networks(self):
+        nets = get_inxi_key(self.inxi, "Network") or []
+        networks = [(nets[i], nets[i + 1]) for i in range(0, len(nets) - 1, 2)]
+
+        for n, iface in networks:
+            model = get_inxi(n, "Device")
+            if not model:
+                continue
+
+            interface = ''
+            for k in n.keys():
+                if "port" in k:
+                    interface = "Integrated"
+                if "pcie" in k:
+                    interface = "PciExpress"
+                if get_inxi(n, "type") == "USB":
+                    interface = "USB"
+
+            self.components.append(
+                {
+                    "type": "NetworkAdapter",
+                    "model": model,
+                    "manufacturer": get_inxi(n, 'vendor'),
+                    "serialNumber": get_inxi(iface, 'mac'),
+                    "speed": get_inxi(n, "speed"),
+                    "interface": interface,
+                }
+            )
+
+    def get_sound_card(self):
+        audio = get_inxi_key(self.inxi, "Audio") or []
+
+        for c in audio:
+            model = get_inxi(c, "Device")
+            if not model:
+                continue
+
+            self.components.append(
+                {
+                    "type": "SoundCard",
+                    "model": model,
+                    "manufacturer": get_inxi(c, 'vendor'),
+                    "serialNumber": get_inxi(c, 'serial'),
+                }
+            )
+
+    def get_display(self):
+        graphics = get_inxi_key(self.inxi, "Graphics") or []
+        for c in graphics:
+            if not get_inxi(c, "Monitor"):
+                continue
+
+            self.components.append(
+                {
+                    "type": "Display",
+                    "model": get_inxi(c, "model"),
+                    "manufacturer": get_inxi(c, "vendor"),
+                    "serialNumber": get_inxi(c, "serial"),
+                    'size': get_inxi(c, "size"),
+                    'diagonal': get_inxi(c, "diag"),
+                    'resolution': get_inxi(c, "res"),
+                    "date": get_inxi(c, "built"),
+                    'ratio': get_inxi(c, "ratio"),
+                }
+            )
+
+    def get_usb_num(self):
+        return len(
+            [
+                u
+                for u in self.dmi.get("Port Connector")
+                if "USB" in u.get("Port Type", "").upper()
+            ]
+        )
+
+    def get_serial_num(self):
+        return len(
+            [
+                u
+                for u in self.dmi.get("Port Connector")
+                if "SERIAL" in u.get("Port Type", "").upper()
+            ]
+        )
+
+    def get_firmware_num(self):
+        return len(
+            [
+                u
+                for u in self.dmi.get("Port Connector")
+                if "FIRMWARE" in u.get("Port Type", "").upper()
+            ]
+        )
+
+    def get_pcmcia_num(self):
+        return len(
+            [
+                u
+                for u in self.dmi.get("Port Connector")
+                if "PCMCIA" in u.get("Port Type", "").upper()
+            ]
+        )
+
+    def get_bios_version(self):
+        return self.dmi.get("BIOS")[0].get("BIOS Revision", '1')
+
+    def loads(self, x):
+        if isinstance(x, str):
+            try:
+                return json.loads(x)
+            except Exception as ss:
+                logger.warning("%s", ss)
+                return {}
+        return x
+
+    def errors(self, txt=None):
+        if not txt:
+            return self._errors
+
+        logger.error(txt)
+        self._errors.append("%s", txt)

evidence/old_parse.py

@@ -0,0 +1,22 @@
+import logging
+
+from evidence.mixin_parse import BuildMix
+
+
+logger = logging.getLogger('django')
+
+
+class Build(BuildMix):
+    # This parse is for get info from snapshots created with old workbench
+    # normaly is worbench 11
+
+    def get_details(self):
+        device = self.json.get('device', {})
+        self.manufacturer = device.get("manufacturer", '')
+        self.model = device.get("model", '')
+        self.chassis = device.get("chassis", '')
+        self.serial_number = device.get("serialNumber", '')
+        self.sku = device.get("sku", '')
+
+    def _get_components(self):
+        self.components = self.json.get("components", [])

evidence/old_parse_details.py

@@ -0,0 +1,13 @@
+import logging
+
+
+logger = logging.getLogger('django')
+
+
+class ParseSnapshot:
+    def __init__(self, snapshot, default="n/a"):
+        self.default = default
+        self.snapshot_json = snapshot
+
+        self.device = snapshot.get("device")
+        self.components = snapshot.get("components")

evidence/parse.py

@@ -3,11 +3,13 @@ import hashlib
 import logging
 
 from evidence import legacy_parse
+from evidence import old_parse
+from evidence import normal_parse
 from evidence.parse_details import ParseSnapshot
 
 from evidence.models import Annotation
 from evidence.xapian import index
-from evidence.parse_details import get_inxi_key, get_inxi
+from evidence.normal_parse_details import get_inxi_key, get_inxi
 from django.conf import settings
 
 if settings.DPP:
@@ -24,8 +26,77 @@ def get_mac(inxi):
         if get_inxi(n, "port"):
             return get_inxi(iface, 'mac')
 
-
 class Build:
+    def __init__(self, evidence_json, user, check=False):
+        """
+        This Build do the save in xapian as document, in Annotations and do
+        register in dlt if is configured for that.
+
+        We have 4 cases for parser diferents snapshots than come from workbench.
+        1) worbench 11 is old_parse.
+        2) legacy is the worbench-script when create a snapshot for devicehub-teal
+        3) some snapshots come as a credential. In this case is parsed as normal_parse
+        4) normal snapshot from worbench-script is the most basic and is parsed as normal_parse
+        """
+        self.evidence = evidence_json.copy()
+        self.uuid = self.evidence['uuid']
+        self.user = user
+
+        if evidence_json.get("credentialSubject"):
+            self.build = normal_parse.Build(evidence_json)
+        elif evidence_json.get("software") != "workbench-script":
+            self.build = old_parse.Build(evidence_json)
+        elif evidence_json.get("data",{}).get("lshw"):
+            self.build = legacy_parse.Build(evidence_json)
+        else:
+            self.build = normal_parse.Build(evidence_json)
+
+        if check:
+            return
+
+        self.index()
+        self.create_annotations()
+        if settings.DPP:
+            self.register_device_dlt()
+
+    def index(self):
+        snap = json.dumps(self.evidence)
+        index(self.user.institution, self.uuid, snap)
+
+    def create_annotations(self):
+        annotation = Annotation.objects.filter(
+                uuid=self.uuid,
+                owner=self.user.institution,
+                type=Annotation.Type.SYSTEM,
+        )
+
+        if annotation:
+            txt = "Warning: Snapshot %s already registered (annotation exists)"
+            logger.warning(txt, self.uuid)
+            return
+
+        for k, v in self.build.algorithms.items():
+            Annotation.objects.create(
+                uuid=self.uuid,
+                owner=self.user.institution,
+                user=self.user,
+                type=Annotation.Type.SYSTEM,
+                key=k,
+                value=self.sign(v)
+            )
+
+    def sign(self, doc):
+        return hashlib.sha3_256(doc.encode()).hexdigest()
+
+    def register_device_dlt(self):
+        legacy_dpp = self.build.algorithms.get('legacy_dpp')
+        chid = self.sign(legacy_dpp)
+        phid = self.sign(json.dumps(self.build.get_doc()))
+        register_device_dlt(chid, phid, self.uuid, self.user)
+        register_passport_dlt(chid, phid, self.uuid, self.user)
+
+
+class Build2:
     def __init__(self, evidence_json, user, check=False):
         if evidence_json.get("data",{}).get("lshw"):
             if evidence_json.get("software") == "workbench-script":

evidence/parse_details.py

@@ -1,408 +1,38 @@
-import json
 import logging
 
-from dmidecode import DMIParse
+from evidence import (
-
+    legacy_parse_details,
-from evidence import legacy_parse_details
+    normal_parse_details,
+    old_parse_details
+)
 
 
 logger = logging.getLogger('django')
 
 
-def get_inxi_key(inxi, component):
-    for n in inxi:
-        for k, v in n.items():
-            if component in k:
-                return v
-
-
-def get_inxi(n, name):
-    for k, v in n.items():
-        if f"#{name}" in k:
-            return v
-
-    return ""
-
-
 class ParseSnapshot:
     def __init__(self, snapshot, default="n/a"):
-        if snapshot.get("data",{}).get("lshw"):
+       if snapshot.get("credentialSubject"):
-            if snapshot.get("software") == "workbench-script":
+           self.build = normal_parse_details.ParseSnapshot(
-                return legacy_parse_details.ParseSnapshot(snapshot, default=default)
+               snapshot,
+               default=default
+           )
+       elif snapshot.get("software") != "workbench-script":
+           self.build = old_parse_details.ParseSnapshot(
+               snapshot,
+               default=default
+           )
+       elif snapshot.get("data",{}).get("lshw"):
+           self.build = legacy_parse_details.ParseSnapshot(
+               snapshot,
+               default=default
+           )
+       else:
+           self.build = normal_parse_details.ParseSnapshot(
+               snapshot,
+               default=default
+           )
 
        self.default = default
-        self.dmidecode_raw = snapshot.get("data", {}).get("dmidecode", "{}")
+       self.device = self.build.snapshot_json.get("device")
-        self.smart_raw = snapshot.get("data", {}).get("smartctl", [])
+       self.components = self.build.snapshot_json.get("components")
-        self.inxi_raw = snapshot.get("data", {}).get("inxi", "") or ""
-        for ev in snapshot.get("evidence", []):
-            if "dmidecode" == ev.get("operation"):
-                self.dmidecode_raw = ev["output"]
-            if "inxi" == ev.get("operation"):
-                self.inxi_raw = ev["output"]
-            if "smartctl" == ev.get("operation"):
-                self.smart_raw = ev["output"]
-        data = snapshot
-        if snapshot.get("credentialSubject"):
-            data = snapshot["credentialSubject"]
-
-        self.device = {"actions": []}
-        self.components = []
-
-        self.dmi = DMIParse(self.dmidecode_raw)
-        self.smart = self.loads(self.smart_raw)
-        self.inxi = self.loads(self.inxi_raw)
-
-        self.set_computer()
-        self.set_components()
-        self.snapshot_json = {
-            "type": "Snapshot",
-            "device": self.device,
-            "software": data["software"],
-            "components": self.components,
-            "uuid": data['uuid'],
-            "endTime": data["timestamp"],
-            "elapsed": 1,
-        }
-
-    def set_computer(self):
-        machine = get_inxi_key(self.inxi, 'Machine') or []
-        for m in machine:
-            system = get_inxi(m, "System")
-            if system:
-                self.device['manufacturer'] = system
-                self.device['model'] = get_inxi(m, "product")
-                self.device['serialNumber'] = get_inxi(m, "serial")
-                self.device['type'] = get_inxi(m, "Type")
-                self.device['chassis'] = self.device['type']
-                self.device['version'] = get_inxi(m, "v")
-            else:
-                self.device['system_uuid'] = get_inxi(m, "uuid")
-                self.device['sku'] = get_inxi(m, "part-nu")
-
-    def set_components(self):
-        self.get_mother_board()
-        self.get_cpu()
-        self.get_ram()
-        self.get_graphic()
-        self.get_display()
-        self.get_networks()
-        self.get_sound_card()
-        self.get_data_storage()
-        self.get_battery()
-
-    def get_mother_board(self):
-        machine = get_inxi_key(self.inxi, 'Machine') or []
-        mb = {"type": "Motherboard",}
-        for m in machine:
-            bios_date = get_inxi(m, "date")
-            if not bios_date:
-                continue
-            mb["manufacturer"] = get_inxi(m, "Mobo")
-            mb["model"] = get_inxi(m, "model")
-            mb["serialNumber"] = get_inxi(m, "serial")
-            mb["version"] = get_inxi(m, "v")
-            mb["biosDate"] = bios_date
-            mb["biosVersion"] = self.get_bios_version()
-            mb["firewire"]: self.get_firmware_num()
-            mb["pcmcia"]: self.get_pcmcia_num()
-            mb["serial"]: self.get_serial_num()
-            mb["usb"]: self.get_usb_num()
-
-            self.get_ram_slots(mb)
-
-        self.components.append(mb)
-
-    def get_ram_slots(self, mb):
-        memory = get_inxi_key(self.inxi, 'Memory') or []
-        for m in memory:
-            slots = get_inxi(m, "slots")
-            if not slots:
-                continue
-            mb["slots"] = slots
-            mb["ramSlots"] = get_inxi(m, "modules")
-            mb["ramMaxSize"] = get_inxi(m, "capacity")
-
-
-    def get_cpu(self):
-        cpu = get_inxi_key(self.inxi, 'CPU') or []
-        cp = {"type": "Processor"}
-        vulnerabilities = []
-        for c in cpu:
-            base = get_inxi(c, "model")
-            if base:
-                cp["model"] = get_inxi(c, "model")
-                cp["arch"] = get_inxi(c, "arch")
-                cp["bits"] = get_inxi(c, "bits")
-                cp["gen"] = get_inxi(c, "gen")
-                cp["family"] = get_inxi(c, "family")
-                cp["date"] = get_inxi(c, "built")
-                continue
-            des = get_inxi(c, "L1")
-            if des:
-                cp["L1"] = des
-                cp["L2"] = get_inxi(c, "L2")
-                cp["L3"] = get_inxi(c, "L3")
-                cp["cpus"] = get_inxi(c, "cpus")
-                cp["cores"] = get_inxi(c, "cores")
-                cp["threads"] = get_inxi(c, "threads")
-                continue
-            bogo = get_inxi(c, "bogomips")
-            if bogo:
-                cp["bogomips"] = bogo
-                cp["base/boost"] = get_inxi(c, "base/boost")
-                cp["min/max"] = get_inxi(c, "min/max")
-                cp["ext-clock"] = get_inxi(c, "ext-clock")
-                cp["volts"] = get_inxi(c, "volts")
-                continue
-            ctype = get_inxi(c, "Type")
-            if ctype:
-                v = {"Type": ctype}
-                status = get_inxi(c, "status")
-                if status:
-                    v["status"] = status
-                mitigation = get_inxi(c, "mitigation")
-                if mitigation:
-                    v["mitigation"] = mitigation
-                vulnerabilities.append(v)
-
-        self.components.append(cp)
-
-
-    def get_ram(self):
-        memory = get_inxi_key(self.inxi, 'Memory') or []
-        mem = {"type": "RamModule"}
-
-        for m in memory:
-            base = get_inxi(m, "System RAM")
-            if base:
-                mem["size"] = get_inxi(m, "total")
-            slot = get_inxi(m, "manufacturer")
-            if slot:
-                mem["manufacturer"] = slot
-                mem["model"] = get_inxi(m, "part-no")
-                mem["serialNumber"] = get_inxi(m, "serial")
-                mem["speed"] = get_inxi(m, "speed")
-                mem["bits"] = get_inxi(m, "data")
-                mem["interface"] = get_inxi(m, "type")
-            module = get_inxi(m, "modules")
-            if module:
-                mem["modules"] = module
-
-        self.components.append(mem)
-
-    def get_graphic(self):
-        graphics = get_inxi_key(self.inxi, 'Graphics') or []
-
-        for c in graphics:
-            if not get_inxi(c, "Device") or not get_inxi(c, "vendor"):
-                continue
-
-            self.components.append(
-                {
-                    "type": "GraphicCard",
-                    "memory": self.get_memory_video(c),
-                    "manufacturer": get_inxi(c, "vendor"),
-                    "model": get_inxi(c, "Device"),
-                    "arch": get_inxi(c, "arch"),
-                    "serialNumber": get_inxi(c, "serial"),
-                    "integrated": True if get_inxi(c, "port") else False
-                }
-            )
-
-    def get_battery(self):
-        bats = get_inxi_key(self.inxi, 'Battery') or []
-        for b in bats:
-            self.components.append(
-                {
-                    "type": "Battery",
-                    "model": get_inxi(b, "model"),
-                    "serialNumber": get_inxi(b, "serial"),
-                    "condition": get_inxi(b, "condition"),
-                    "cycles": get_inxi(b, "cycles"),
-                    "volts": get_inxi(b, "volts")
-                }
-            )
-
-    def get_memory_video(self, c):
-        memory = get_inxi_key(self.inxi, 'Memory') or []
-
-        for m in memory:
-            igpu = get_inxi(m, "igpu")
-            agpu = get_inxi(m, "agpu")
-            ngpu = get_inxi(m, "ngpu")
-            gpu = get_inxi(m, "gpu")
-            if igpu or agpu or gpu or ngpu:
-                return igpu or agpu or gpu or ngpu
-
-        return self.default
-
-    def get_data_storage(self):
-        hdds= get_inxi_key(self.inxi, 'Drives') or []
-        for d in hdds:
-            usb = get_inxi(d, "type")
-            if usb == "USB":
-                continue
-
-            serial = get_inxi(d, "serial")
-            if serial:
-                hd = {
-                    "type": "Storage",
-                    "manufacturer": get_inxi(d, "vendor"),
-                    "model": get_inxi(d, "model"),
-                    "serialNumber": get_inxi(d, "serial"),
-                    "size": get_inxi(d, "size"),
-                    "speed": get_inxi(d, "speed"),
-                    "interface": get_inxi(d, "tech"),
-                    "firmware": get_inxi(d, "fw-rev")
-                }
-                rpm = get_inxi(d, "rpm")
-                if rpm:
-                    hd["rpm"] = rpm
-
-                family = get_inxi(d, "family")
-                if family:
-                    hd["family"] = family
-
-                sata = get_inxi(d, "sata")
-                if sata:
-                    hd["sata"] = sata
-
-                continue
-
-
-            cycles = get_inxi(d, "cycles")
-            if cycles:
-                hd['cycles'] = cycles
-                hd["health"] = get_inxi(d, "health")
-                hd["time of used"] = get_inxi(d, "on")
-                hd["read used"] = get_inxi(d, "read-units")
-                hd["written used"] = get_inxi(d, "written-units")
-
-                self.components.append(hd)
-                continue
-
-            hd = {}
-
-    def sanitize(self, action):
-        return []
-
-    def get_networks(self):
-        nets = get_inxi_key(self.inxi, "Network") or []
-        networks = [(nets[i], nets[i + 1]) for i in range(0, len(nets) - 1, 2)]
-
-        for n, iface in networks:
-            model = get_inxi(n, "Device")
-            if not model:
-                continue
-
-            interface = ''
-            for k in n.keys():
-                if "port" in k:
-                    interface = "Integrated"
-                if "pcie" in k:
-                    interface = "PciExpress"
-                if get_inxi(n, "type") == "USB":
-                    interface = "USB"
-
-            self.components.append(
-                {
-                    "type": "NetworkAdapter",
-                    "model": model,
-                    "manufacturer": get_inxi(n, 'vendor'),
-                    "serialNumber": get_inxi(iface, 'mac'),
-                    "speed": get_inxi(n, "speed"),
-                    "interface": interface,
-                }
-            )
-
-    def get_sound_card(self):
-        audio = get_inxi_key(self.inxi, "Audio") or []
-
-        for c in audio:
-            model = get_inxi(c, "Device")
-            if not model:
-                continue
-
-            self.components.append(
-                {
-                    "type": "SoundCard",
-                    "model": model,
-                    "manufacturer": get_inxi(c, 'vendor'),
-                    "serialNumber": get_inxi(c, 'serial'),
-                }
-            )
-
-    def get_display(self):
-        graphics = get_inxi_key(self.inxi, "Graphics") or []
-        for c in graphics:
-            if not get_inxi(c, "Monitor"):
-                continue
-
-            self.components.append(
-                {
-                    "type": "Display",
-                    "model": get_inxi(c, "model"),
-                    "manufacturer": get_inxi(c, "vendor"),
-                    "serialNumber": get_inxi(c, "serial"),
-                    'size': get_inxi(c, "size"),
-                    'diagonal': get_inxi(c, "diag"),
-                    'resolution': get_inxi(c, "res"),
-                    "date": get_inxi(c, "built"),
-                    'ratio': get_inxi(c, "ratio"),
-                }
-            )
-
-    def get_usb_num(self):
-        return len(
-            [
-                u
-                for u in self.dmi.get("Port Connector")
-                if "USB" in u.get("Port Type", "").upper()
-            ]
-        )
-
-    def get_serial_num(self):
-        return len(
-            [
-                u
-                for u in self.dmi.get("Port Connector")
-                if "SERIAL" in u.get("Port Type", "").upper()
-            ]
-        )
-
-    def get_firmware_num(self):
-        return len(
-            [
-                u
-                for u in self.dmi.get("Port Connector")
-                if "FIRMWARE" in u.get("Port Type", "").upper()
-            ]
-        )
-
-    def get_pcmcia_num(self):
-        return len(
-            [
-                u
-                for u in self.dmi.get("Port Connector")
-                if "PCMCIA" in u.get("Port Type", "").upper()
-            ]
-        )
-
-    def get_bios_version(self):
-        return self.dmi.get("BIOS")[0].get("BIOS Revision", '1')
-
-    def loads(self, x):
-        if isinstance(x, str):
-            try:
-                return json.loads(x)
-            except Exception as ss:
-                logger.warning("%s", ss)
-                return {}
-        return x
-
-    def errors(self, txt=None):
-        if not txt:
-            return self._errors
-
-        logger.error(txt)
-        self._errors.append("%s", txt)

utils/constants.py

@@ -13,7 +13,7 @@ HID_ALGO1 = [
     "manufacturer",
     "model",
     "chassis",
-    "serialNumber",
+    "serial_number",
     "sku"
 ]
 
@@ -21,7 +21,7 @@ LEGACY_DPP = [
     "manufacturer",
     "model",
     "chassis",
-    "serialNumber",
+    "serial_number",
     "sku",
     "type",
     "version"