FYP Updates: UI Improvements, ESP32 Support, Pipeline Integration and Documentation

.gitignore

@@ -87,3 +87,13 @@ routes.py.bak
 backend/__pycache__/
 *.pyc
 unblob.log
+data/
+backend/venv/
+backend/__pycache__/
+backend/core/**/__pycache__/
+*.pyc
+*.pyo
+frontend/node_modules/
+
+# Reference dashboard template source
+corona-free-dark-bootstrap-admin-template-1.0.0/

README.md

@@ -15,6 +15,19 @@
 **A full-stack firmware security assessment platform for IoT devices.**  
 Automated 7-phase pipeline: device detection → firmware extraction → static analysis → SBOM/HBOM/CBOM → CVE mapping → VEX risk triage → compliance report (PDF).
 
+<p align="center">
+  <strong>Supervisor</strong><br>
+  Dr. Tahir Mushtaq
+</p>
+
+<p align="center">
+  <strong>Collaborators</strong><br>
+  Imran Hashmi &nbsp;•&nbsp;
+  Muhammad Moeed &nbsp;•&nbsp;
+  Muhammad Hassan &nbsp;•&nbsp;
+  Zain-Ul-Abedin
+</p>
+
 [Features](#-features) · [Architecture](#-architecture) · [Quick Start](#-quick-start) · [Pipeline](#-7-phase-pipeline) · [Hardware](#-hardware-support) · [Status](#-project-status)
 
 </div>

backend/core/compliance/report_generator.py

@@ -17,7 +17,7 @@
 
 logger = logging.getLogger(__name__)
 
-OUTPUT_BASE = Path("/tmp/iifvdcas_compliance")
+OUTPUT_BASE = Path("/mnt/d/Uni Work/FYP 2 IOT/IIFVDCAS-main/IIFVDCAS-main/data/compliance")
 
 
 @dataclass
@@ -98,7 +98,7 @@ def _load_sbom(sbom_job_id: str) -> Dict:
     except Exception:
         pass
     # Disk fallback
-    base = Path(f"/tmp/iifvdcas_sbom/{sbom_job_id}")
+    base = Path(f"/mnt/d/Uni Work/FYP 2 IOT/IIFVDCAS-main/IIFVDCAS-main/data/sbom/{sbom_job_id}")
     result = {}
     for fname, key in [("cyclonedx_sbom.json","sbom"),
                        ("hbom.json","hbom"), ("cbom.json","cbom")]:

backend/core/cve_mapping/build_nvd_index.py

@@ -2,7 +2,7 @@
 """
 Build a SQLite index of the local NVD per-CVE JSON feed.
 Run once: python3 build_nvd_index.py
-Output:  /tmp/iifvdcas_nvd.db  (~50–80 MB)
+Output:  /mnt/d/Uni Work/FYP 2 IOT/IIFVDCAS-main/IIFVDCAS-main/data/nvd.db  (~50–80 MB)
 
 Schema:
   cve(id, published, cvss_score, cvss_severity, cvss_vector, description, cwe_ids, path)
@@ -16,7 +16,7 @@
 
 NVD_ROOT  = Path.home() / "emba" / "external" / "nvd-json-data-feeds"
 EPSS_DIR  = Path.home() / "emba" / "external" / "EPSS-data" / "EPSS_CVE_data"
-DB_PATH   = Path("/tmp/iifvdcas_nvd.db")
+DB_PATH   = Path("/mnt/d/Uni Work/FYP 2 IOT/IIFVDCAS-main/IIFVDCAS-main/data/nvd.db")
 STATE_CSV = NVD_ROOT / "_state.csv"
 
 def get_cve_path(cve_id: str) -> Path:

backend/core/cve_mapping/cve_aggregator.py

@@ -109,7 +109,7 @@ def _load_components(sbom_job_id: str) -> List[Dict]:
     except Exception as e:
         logger.debug(f"[CVE] memory lookup: {e}")
 
-    base = Path(f"/tmp/iifvdcas_sbom/{sbom_job_id}")
+    base = Path(f"/mnt/d/Uni Work/FYP 2 IOT/IIFVDCAS-main/IIFVDCAS-main/data/sbom/{sbom_job_id}")
     for name in ("cyclonedx_sbom.json", "sbom.json", "components.json", "bom.json"):
         fp = base / name
         if not fp.exists():
@@ -152,7 +152,7 @@ def _is_real_package(comp: Dict) -> bool:
     """Return True only if this component should be looked up in NVD."""
     name   = (comp.get("name")        or "").lower().strip()
     source = (comp.get("source_path") or "").lower()
-    vendor = (comp.get("vendor")      or "").strip()
+    vendor = (comp.get("vendor") or comp.get("supplier") or "").strip()
     cpe    = (comp.get("cpe")         or "").strip()
 
     if name in _SECURITY_MARKERS:
@@ -168,7 +168,7 @@ def _is_real_package(comp: Dict) -> bool:
 
 def _should_do_product_lookup(comp: Dict) -> bool:
     """Return True if a name-based NVD product lookup is appropriate."""
-    vendor = (comp.get("vendor")      or "").strip()
+    vendor = (comp.get("vendor") or comp.get("supplier") or "").strip()
     source = (comp.get("source_path") or "").lower()
     if source in _FIRMWARE_ID_SOURCES:
         return False
@@ -202,7 +202,7 @@ def _map_all(
         name    = (comp.get("name")    or "").strip()
         version = (comp.get("version") or "").strip()
         cpe     = (comp.get("cpe")     or "").strip()
-        vendor  = (comp.get("vendor")  or "").strip()
+        vendor  = (comp.get("vendor") or comp.get("supplier") or "").strip()
 
         pct = 10 + int(i / max(total, 1) * 55)
         emit(pct, f"[{i+1}/{total}] {name}")

backend/core/cve_mapping/epss_scorer.py

@@ -9,7 +9,7 @@
 
 logger = logging.getLogger(__name__)
 
-DB_PATH  = Path("/tmp/iifvdcas_nvd.db")
+DB_PATH  = Path("/mnt/d/Uni Work/FYP 2 IOT/IIFVDCAS-main/IIFVDCAS-main/data/nvd.db")
 EPSS_DIR = Path.home() / "emba" / "external" / "EPSS-data" / "EPSS_CVE_data"
 
 _csv_cache: Dict[str, float] = {}

backend/core/cve_mapping/kev_checker.py

@@ -14,7 +14,7 @@
 
 # KEV data locations
 KEV_URL = "https://www.cisa.gov/sites/default/files/feeds/known_exploited_vulnerabilities.json"
-KEV_CACHE_PATH = Path("/tmp/iifvdcas_kev_cache.json")
+KEV_CACHE_PATH = Path("/mnt/d/Uni Work/FYP 2 IOT/IIFVDCAS-main/IIFVDCAS-main/data/kev_cache.json")
 KEV_CACHE_TTL = 86400  # 24 hours in seconds
 
 # In-memory cache for KEV IDs

backend/core/cve_mapping/nvd_lookup.py

@@ -16,7 +16,7 @@
 
 logger = logging.getLogger(__name__)
 
-DB_PATH     = Path("/tmp/iifvdcas_nvd.db")
+DB_PATH     = Path("/mnt/d/Uni Work/FYP 2 IOT/IIFVDCAS-main/IIFVDCAS-main/data/nvd.db")
 NVD_API_URL = "https://services.nvd.nist.gov/rest/json/cves/2.0"
 OSV_BATCH   = "https://api.osv.dev/v1/querybatch"
 

backend/core/firmware_extraction/extractor.py

@@ -25,7 +25,7 @@
 
 logger = logging.getLogger(__name__)
 
-EXTRACT_BASE = Path("/tmp/iifvdcas_extractions")
+EXTRACT_BASE = Path("/mnt/d/Uni Work/FYP 2 IOT/IIFVDCAS-main/IIFVDCAS-main/data/extractions")
 
 
 # ─── Enums matching EMBA detection logic ─────────────────────────────────────
@@ -256,6 +256,9 @@ def analyze_firmware_file(fw_path: str) -> FirmwareFileInfo:
         info.firmware_type = FirmwareType.UBOOT
     elif "CPIO ARCHIVE" in f:
         info.firmware_type = FirmwareType.CPIO
+    elif "ESP IMAGE" in bw or "ESP32" in bw or "XTENSA" in bw.upper():
+        info.firmware_type = FirmwareType.RAW_BINARY
+        info.architecture_hint = "Xtensa"
     elif any(k in f for k in ["GZIP", "ZIP", "XZ", "BZIP2", "7-ZIP", "TAR", "ISO 9660"]):
         info.firmware_type = FirmwareType.ARCHIVE
         info.is_compressed = True
@@ -636,7 +639,7 @@ def progress(pct: int, step: str):
         status=ExtractionStatus.RUNNING,
     )
 
-    # Output dir: /tmp/iifvdcas_extractions/<job_id>/
+    # Output dir: /mnt/d/Uni Work/FYP 2 IOT/IIFVDCAS-main/IIFVDCAS-main/data/extractions/<job_id>/
     output_dir = str(EXTRACT_BASE / job_id)
     os.makedirs(output_dir, exist_ok=True)
     result.output_dir = output_dir
@@ -745,10 +748,17 @@ def progress(pct: int, step: str):
                         log("[P50] binwalk extraction succeeded")
                         progress(50, "binwalk extraction complete")
                     else:
-                        log("[P60] Both unblob and binwalk found nothing — firmware may be raw binary")
-                        result.extractor_used = "none"
-                        result.status = ExtractionStatus.PARTIAL
-                        progress(50, "No extractor succeeded")
+                        log("[P60] Both unblob and binwalk found nothing — copying raw binary for string analysis")
+                        # Copy raw firmware to output dir so S108 can scan it
+                        import shutil
+                        raw_dir = os.path.join(output_dir, "raw_binary")
+                        os.makedirs(raw_dir, exist_ok=True)
+                        raw_copy = os.path.join(raw_dir, Path(fw_path).name)
+                        shutil.copy2(fw_path, raw_copy)
+                        extracted = True
+                        result.extractor_used = "raw_copy"
+                        log(f"[P60] Raw binary copied to {raw_copy} for string-based analysis")
+                        progress(50, "Raw binary ready for analysis")
 
         # ── Step 3: Count extracted files ─────────────────────────────────
         progress(60, "Counting extracted files")

backend/core/firmware_extraction/routes.py

@@ -15,6 +15,7 @@
 import re
 import shutil
 import subprocess
+import sys
 import threading
 from datetime import datetime, timezone
 from pathlib import Path
@@ -30,8 +31,8 @@
 
 router = APIRouter(prefix="/api/v1/firmware", tags=["firmware"])
 
-UPLOAD_DIR    = Path("/tmp/iifvdcas_uploads");    UPLOAD_DIR.mkdir(parents=True, exist_ok=True)
-FIRMWARE_STORE = Path("/tmp/iifvdcas_firmware_store"); FIRMWARE_STORE.mkdir(parents=True, exist_ok=True)
+UPLOAD_DIR    = Path("/mnt/d/Uni Work/FYP 2 IOT/IIFVDCAS-main/IIFVDCAS-main/data/uploads");    UPLOAD_DIR.mkdir(parents=True, exist_ok=True)
+FIRMWARE_STORE = Path("/mnt/d/Uni Work/FYP 2 IOT/IIFVDCAS-main/IIFVDCAS-main/data/firmware_store"); FIRMWARE_STORE.mkdir(parents=True, exist_ok=True)
 
 extraction_jobs: dict[str, dict] = {}
 
@@ -199,26 +200,29 @@ def _esptool_read_cmd(
     no_stub:     bool = False,
 ) -> list[str]:
     """
-    Build the correct esptool read_flash command.
+    Build the correct esptool v5 read-flash command.
 
     Key fixes vs previous version:
-      - 'read_flash' (underscore), NOT 'read-flash' or 'read-flash'
+      - 'read-flash' (hyphen), as required by esptool v5
       - Flash size is an explicit hex integer, NOT the string "ALL"
-      - --before default_reset (underscore), --after hard_reset (underscore)
+      - --before default-reset, --after hard-reset
       - --no-stub flag added when no_stub=True
     """
     size_arg = hex(flash_bytes) if flash_bytes else "0x400000"  # default 4 MB
+    # Use the API process's interpreter so virtual-environment PATH handling
+    # cannot make the installed esptool command disappear.
     cmd = [
-        "esptool",
+        sys.executable,
+        "-m", "esptool",
         "--port",   device_port,
         "--baud",   str(baud),
         "--chip",   chip,
-        "--before", "default_reset",
-        "--after",  "hard_reset",
+        "--before", "default-reset",
+        "--after",  "hard-reset",
     ]
     if no_stub:
         cmd.append("--no-stub")
-    cmd += ["read_flash", "0x0", size_arg, out_path]
+    cmd += ["read-flash", "0x0", size_arg, out_path]
     return cmd
 
 
@@ -247,7 +251,7 @@ async def emit(event: str, **kwargs):
     await emit("extraction_started")
     await emit("progress", percent=2, step="Preparing directories")
 
-    output_dir = Path("/tmp/iifvdcas_extractions") / job_id
+    output_dir = Path("/mnt/d/Uni Work/FYP 2 IOT/IIFVDCAS-main/IIFVDCAS-main/data/extractions") / job_id
     output_dir.mkdir(parents=True, exist_ok=True)
     ts          = datetime.now(timezone.utc).strftime("%Y%m%d_%H%M%S")
     chip_label  = chip if tool == "esptool" else mcu
@@ -346,6 +350,7 @@ def _stream(cmd=cmd, timeout_s=timeout_s):
         attempt_error = ""
         corrupt_data  = False
         detected_size = None
+        last_read_pct = -1
 
         while True:
             try:
@@ -367,9 +372,12 @@ def _stream(cmd=cmd, timeout_s=timeout_s):
                     continue
 
                 # Parse progress percentage
-                m_pct = re.search(r"\(\s*(\d+)\s*%\)", value)
+                m_pct = re.search(r"(\d+(?:\.\d+)?)%", value)
                 if m_pct:
-                    pct = int(m_pct.group(1))
+                    pct = int(float(m_pct.group(1)))
+                    if pct == last_read_pct:
+                        continue
+                    last_read_pct = pct
                     stub_label = "(--no-stub)" if is_no_stub else ""
                     await emit("progress",
                                percent=5 + int(pct * 0.15),
@@ -553,6 +561,17 @@ async def get_file_tree(job_id: str):
             "flash_bin_size_human": job.get("flash_bin_size_human")}
 
 
+@router.get("/extract/{job_id}/download")
+async def download_firmware(job_id: str):
+    if job_id not in extraction_jobs:
+        raise HTTPException(404, "Job not found")
+    job = extraction_jobs[job_id]
+    path = job.get("flash_bin_path")
+    filename = job.get("flash_bin_filename", "flash.bin")
+    if not path or not Path(path).exists():
+        raise HTTPException(404, "Firmware file not found")
+    return FileResponse(path, media_type="application/octet-stream", filename=filename)
+
 @router.delete("/extract/{job_id}")
 async def delete_extraction(job_id: str):
     cleanup_extraction(job_id)

backend/core/sbom/avr_component_extractor.py

@@ -64,8 +64,17 @@
      None, None,     None),
     ("ESP8266",      "esp8266",       "espressif",    "esp8266_sdk",
      None, None,     None),
+    # MicroPython (must come before ESP32 generic)
+    ("MicroPython",  "micropython",   "micropython",  "micropython",
+     None, None,     r"MicroPython[\s/v]+([\d\.]+)"),
+    ("MPY version",  "micropython",   "micropython",  "micropython",
+     None, None,     r"MPY version\s*:\s*v?([\d\.]+)"),
+
+    # ESP-IDF framework
+    ("ESP-IDF",      "esp-idf",       "espressif",    "esp-idf",
+     None, None,     r"IDF version\s*:\s*v?([\d\.]+)"),
     ("ESP32",        "esp32",         "espressif",    "esp-idf",
-     None, None,     None),
+     None, None,     r"ESP-IDF[:\s/v]+([\d\.]+)"),
 
     # ── Parsing / Serialization ───────────────────────────────────────────────
     ("ArduinoJson",  "arduinojson",   "bblanchon",    "arduinojson",
@@ -225,6 +234,18 @@ def detect_avr_components(
     # Firmware identity
     fw_name = "custom-avr-firmware"
     fw_ver  = "unknown"
+    # MicroPython specific version extraction
+    for s in all_strings:
+        m = re.search(r'MicroPython[\s/v]+([\d\.]+)', s, re.IGNORECASE)
+        if m:
+            fw_name = "micropython"
+            fw_ver = m.group(1)
+            break
+        m2 = re.search(r'MPY version\s*:\s*v?([\d\.]+)', s)
+        if m2:
+            fw_name = "micropython"
+            fw_ver = m2.group(1)
+            break
     for s in all_strings:
         m = re.search(r'([A-Za-z][A-Za-z0-9\-]{3,20})[-_]IoT|([A-Za-z][A-Za-z0-9\-]{3,20})-Firmware', s)
         if m:

backend/core/sbom/generator.py

@@ -217,14 +217,20 @@ def run_f02_toolchain(target_dir: str, log_cb) -> dict:
     bin_files = list(Path(target_dir).rglob("*.bin"))
     for bf in bin_files[:3]:
         str_out, _, _ = _run(["strings", "-n", "4", str(bf)], timeout=10)
-        if "avr-gcc" in str_out.lower() or "avr" in str_out.lower():
+        strings_lower = str_out.lower()
+        if any(marker in strings_lower for marker in ("esp-idf", "esp32", "esp8266", "xtensa")):
+            toolchain["architecture"] = "Xtensa (ESP32/ESP8266)"
+            toolchain["compiler_hint"] = "ESP-IDF"
+            toolchain["libc_type"] = "newlib"
+            break
+        if "avr-gcc" in strings_lower or "avr-libc" in strings_lower:
             toolchain["architecture"] = "AVR (ATmega)"
             toolchain["compiler_hint"] = "avr-gcc"
             toolchain["libc_type"] = "avr-libc"
             break
     # If still unknown and .bin/.hex files present → AVR
     if toolchain["architecture"] == "unknown":
-        if any(Path(target_dir).rglob("*.bin")) or any(Path(target_dir).rglob("*.hex")):
+        if any(Path(target_dir).rglob("*.hex")):
             toolchain["architecture"] = "AVR (ATmega)"
             toolchain["libc_type"] = "avr-libc"
     log_cb(f"[F02] Arch: {toolchain['architecture']} | libc: {toolchain['libc_type']} | compiler: {toolchain['compiler_hint']}")
@@ -525,7 +531,7 @@ def progress(pct: int, step: str):
         status="running",
     )
     prior = prior_analysis or {}
-    output_dir = Path(f"/tmp/iifvdcas_sbom/{job_id}")
+    output_dir = Path(f"/mnt/d/Uni Work/FYP 2 IOT/IIFVDCAS-main/IIFVDCAS-main/data/sbom/{job_id}")
     output_dir.mkdir(parents=True, exist_ok=True)
 
     if not os.path.isdir(target_dir):
@@ -546,13 +552,31 @@ def progress(pct: int, step: str):
         cfg_comps = _detect_config_components(target_dir, log)
         all_comps = elf_comps + cfg_comps
 
+        if result.toolchain.get("architecture", "").startswith("Xtensa"):
+            esp_strings = ""
+            for bf in list(Path(target_dir).rglob("*.bin"))[:3]:
+                out, _, _ = _run(["strings", "-n", "4", str(bf)], timeout=15)
+                esp_strings += out + "\n"
+            match = re.search(r"(?:ESP-IDF[^\n]{0,80}|[A-Za-z0-9_-]*idf)[^\n]*?v?(\d+\.\d+\.\d+)", esp_strings, re.I)
+            if not match:
+                match = re.search(r"(?:idf|esp|qa-test)[A-Za-z0-9_-]*-v(\d+\.\d+\.\d+)", esp_strings, re.I)
+            if match:
+                version = match.group(1)
+                all_comps.append(SBOMComponent(
+                    bom_ref=str(uuid.uuid4()), component_type="framework",
+                    name="esp-idf", version=version, supplier="espressif",
+                    license="Apache-2.0", source_path="esp-idf-version-scan",
+                    cpe=f"cpe:2.3:a:espressif:esp-idf:{version}:*:*:*:*:*:*:*",
+                    purl=f"pkg:generic/espressif/esp-idf@{version}",
+                ))
+                log(f"[ESP-SBOM] ESP-IDF {version} detected from firmware strings")
+            else:
+                log("[ESP-SBOM] ESP-IDF detected but no reliable version string was found")
+
         # ── AVR/MCU: Embedded library detection ──────────────────────────
         # For raw AVR firmware (no ELF/package managers), scan binary
         # strings for known Arduino/embedded library signatures with CPEs.
-        is_avr = (
-            result.toolchain.get("architecture", "").upper().startswith("AVR")
-            or not elf_comps
-        )
+        is_avr = result.toolchain.get("architecture", "").upper().startswith("AVR")
         if is_avr:
             progress(38, "AVR: Detecting embedded library components")
             try: