pxe-server/webapp/services/blancco_report.py

"""
Parse Blancco Drive Eraser XML report into a friendly Python dict.

Blancco encodes everything as <entries name="X"><entry name="K" type="T">V</entry>...
which is too noisy to read directly in a template. Walking the tree once into
nested dicts/lists lets the template render fields by their semantic name
without any XPath-style noise.

Returned shape (best-effort; missing keys are OK in templates):
    {
      "meta":       {document_id, date, product_name, product_version,
                     product_revision, integrity},
      "company":    {business_location, business_name, customer_license, ...},
      "license_consumption_ids": [str, ...],
      "erasures":   [ {erasure_id, timestamp, target:{...}, state,
                       elapsed_time, erasure_standard_name, step:[...], ...} ],
      "hardware":   {system:{...}, bios:{...}, processors:{...}, ...},
    }
"""
from __future__ import annotations
import os
import xml.etree.ElementTree as ET


def _walk(node):
    # Recursive walk over Blancco's <entry|entries name="K"> children.
    # Repeated names collapse to a list.
    result = {}
    for child in node:
        name = child.attrib.get("name") or child.tag
        if child.tag == "entry":
            val = (child.text or "").strip()
        elif child.tag == "entries":
            val = _walk(child)
        else:
            continue
        if name in result:
            existing = result[name]
            if not isinstance(existing, list):
                result[name] = [existing]
            result[name].append(val)
        else:
            result[name] = val
    return result


def _text(parent, path, default=""):
    if parent is None:
        return default
    el = parent.find(path)
    if el is None or el.text is None:
        return default
    return el.text.strip()


# Parsed reports by path, reused while the file's size and mtime are
# unchanged. The reports list, the to-do (which walks a year of reports
# several times per request), exports and evidence packs all parse the same
# files; a report on the share is never edited in place. Callers must not
# mutate the returned dict.
_parse_cache: dict = {}


def parse(path: str) -> dict:
    st = os.stat(path)
    sig = (st.st_mtime_ns, st.st_size)
    hit = _parse_cache.get(path)
    if hit and hit[0] == sig:
        return hit[1]
    data = _parse(path)
    if len(_parse_cache) > 5000:  # far above a site's report count; just a bound
        _parse_cache.clear()
    _parse_cache[path] = (sig, data)
    return data


def _parse(path: str) -> dict:
    tree = ET.parse(path)
    root = tree.getroot()
    rep = root.find(".//report/blancco_data") or root

    out = {
        "meta": {},
        "company": {},
        "license_consumption_ids": [],
        "erasures": [],
        "hardware": {},
    }

    desc = rep.find("description")
    if desc is not None:
        out["meta"]["document_id"] = _text(desc, "document_id")
        log_entry = desc.find(".//log_entry")
        if log_entry is not None:
            author = log_entry.find("author")
            if author is not None:
                out["meta"]["product_name"]     = _text(author, "product_name")
                out["meta"]["product_version"]  = _text(author, "product_version")
                out["meta"]["product_revision"] = _text(author, "product_revision")
            out["meta"]["date"]      = _text(log_entry, "date")
            out["meta"]["integrity"] = _text(log_entry, "integrity")

        company = desc.find(".//entry[@name='description_entries']/entries[@name='company_information']")
        if company is not None:
            for e in company.findall("entry"):
                k = e.attrib.get("name", "")
                if k:
                    out["company"][k] = (e.text or "").strip()

        lcids = desc.find(".//entry[@name='description_entries']/entries[@name='license_consumption_ids']")
        if lcids is not None:
            for e in lcids.findall("entry"):
                v = (e.text or "").strip()
                if v:
                    out["license_consumption_ids"].append(v)

    for er in rep.findall(".//blancco_erasure_report//entries[@name='erasure']"):
        out["erasures"].append(_walk(er))

    hw = rep.find("blancco_erasure_report/../blancco_hardware_report") or rep.find(".//blancco_hardware_report")
    if hw is not None:
        for sub in hw:
            if sub.tag != "entries":
                continue
            out["hardware"][sub.attrib.get("name", "")] = _walk(sub)

    return out


def report_datetime(data: dict):
    """Report creation time from the report itself, or None if missing.

    Blancco writes ISO 8601 with a numeric offset, e.g.
    2026-09-04T17:00:22+0000. File mtime is not reliable for audit ranges:
    copying reports onto the share resets it.
    """
    from datetime import datetime
    raw = (data.get("meta") or {}).get("date") or ""
    try:
        return datetime.strptime(raw, "%Y-%m-%dT%H:%M:%S%z")
    except ValueError:
        return None


EXPORT_COLUMNS = [
    "report_file", "report_date", "document_id", "product_version",
    "system_manufacturer", "system_model", "system_serial",
    "drive_model", "drive_serial", "drive_capacity_gb", "drive_interface",
    "erasure_standard", "state", "start_time", "end_time", "elapsed_time",
    "total_errors", "failure_reason", "license_consumption_id",
    "business_name", "business_location", "cloud_upload",
]


def _as_list(x):
    if x is None or x == "":
        return []
    return x if isinstance(x, list) else [x]


def failure_reason(er: dict) -> str:
    """Why an erasure did not succeed, from the report itself; "" if it did.

    Blancco records exception messages under erasure_details (e.g.
    "NBP area content was not erased."), a state and error count per step,
    and total/read/mismatch error counts. Auditors (ENT-158/159) want the
    cause of failure next to each failed wipe.
    """
    if (er.get("state") or "").strip().lower() == "successful":
        return ""
    parts = []
    details = er.get("erasure_details") or {}
    for exc in _as_list(details.get("exception")) + _as_list(details.get("messages")):
        msg = exc.get("message") if isinstance(exc, dict) else exc
        if msg and msg not in parts:
            parts.append(msg)
    steps = (er.get("steps") or {}).get("step")
    for i, st in enumerate(_as_list(steps), 1):
        if isinstance(st, dict) and (st.get("state") or "completed") != "completed":
            name = st.get("name") or st.get("step_name") or f"step {i}"
            parts.append(f"{name}: {st.get('state')}")
    for key in ("total_errors", "read_errors", "mismatch_errors"):
        try:
            n = int(er.get(key) or 0)
        except ValueError:
            n = 0
        if n:
            parts.append(f"{key.replace('_', ' ')}: {n}")
    return "; ".join(parts) or f"state: {er.get('state') or 'unknown'}"


def _drive_key(er: dict, index: int) -> str:
    target = er.get("target") or {}
    if isinstance(target, list):
        target = target[0] if target else {}
    return target.get("serial") or f"#{index}"


def drive_outcomes(data: dict) -> dict:
    """Per drive: did any erasure on it succeed?

    Blancco falls back to a weaker standard when one fails on a drive (e.g.
    NIST 800-88 Purge fails, then Clear succeeds) and the report keeps both
    entries. The drive is wiped if any of its erasures succeeded; that is
    what the GE work instruction counts as a successful wipe.
    Returns {drive_key: {"ok": bool, "standard": str}}.
    """
    out = {}
    for i, er in enumerate(data.get("erasures") or []):
        key = _drive_key(er, i)
        ok = (er.get("state") or "").strip().lower() == "successful"
        cur = out.setdefault(key, {"ok": False, "standard": ""})
        if ok and not cur["ok"]:
            cur.update(ok=True, standard=er.get("erasure_standard_name", ""))
    return out


def report_result(data: dict) -> str:
    """"Successful" when every drive was wiped, "Failed" if any drive was
    not, "" when the report has no erasures."""
    outcomes = drive_outcomes(data)
    if not outcomes:
        return ""
    return "Successful" if all(o["ok"] for o in outcomes.values()) else "Failed"


def export_rows(filename: str, data: dict, when: str, cloud: str = "") -> list[dict]:
    """One CSV row per erased drive; a report with no erasures still gets a row.

    cloud is the upload status recorded by the webapp for this report
    ("sent", "failed" or "" when unknown).
    """
    meta = data.get("meta") or {}
    system = (data.get("hardware") or {}).get("system") or {}
    if isinstance(system, list):
        system = system[0] if system else {}
    base = {
        "report_file": filename,
        "report_date": when,
        "document_id": meta.get("document_id", ""),
        "product_version": meta.get("product_version", ""),
        "system_manufacturer": system.get("manufacturer", ""),
        "system_model": system.get("model", ""),
        "system_serial": system.get("serial", ""),
        "license_consumption_id": ";".join(data.get("license_consumption_ids") or []),
        "business_name": (data.get("company") or {}).get("business_name", ""),
        "business_location": (data.get("company") or {}).get("business_location", ""),
        "cloud_upload": cloud,
    }
    rows = []
    outcomes = drive_outcomes(data)
    for i, er in enumerate(data.get("erasures") or []):
        reason = failure_reason(er)
        outcome = outcomes.get(_drive_key(er, i)) or {}
        if reason and outcome.get("ok"):
            reason = f"Fallback: {reason}; then erased with {outcome.get('standard') or 'another standard'}"
        target = er.get("target") or {}
        if isinstance(target, list):
            target = target[0] if target else {}
        try:
            cap = f"{int(target.get('capacity', '')) / 1e9:.1f}"
        except ValueError:
            cap = ""
        rows.append({
            **base,
            "drive_model": target.get("model", ""),
            "drive_serial": target.get("serial", ""),
            "drive_capacity_gb": cap,
            "drive_interface": target.get("interface_type", ""),
            "erasure_standard": er.get("erasure_standard_name", ""),
            "state": er.get("state", ""),
            "start_time": er.get("start_time", ""),
            "end_time": er.get("end_time", ""),
            "elapsed_time": er.get("elapsed_time", ""),
            "total_errors": er.get("total_errors", ""),
            "failure_reason": reason,
        })
    return rows or [base]