020f043d8b
Technician To-Do (sidebar link with a count, and a banner on every page while anything is open) lists the playbook steps the server cannot do: 1. wipes still to tick on the Onsite Bin RITM (ENT-045/159) 2. monthly evidence packs to upload to Box (ENT-056/158), Mark uploaded 3. drives that failed to wipe, to destroy (COD recorded) 4. reports Blancco could not upload to the cloud (Resolved) Marks are stored in the Blancco state file; the counts are cached for 60s and refreshed on every mark, ServiceNow tick and cloud-status post. Results are now judged per drive. Blancco keeps every attempt in the report, so a NIST Purge that failed followed by a successful Clear fallback on the same drive appeared as two entries, and the Reports page rolled the whole report up to Failed. 20 such wipes since May showed as Failed; they were wiped (the work instruction counts the fallback as a success). The CSV marks those rows "Fallback: ... then erased with ...". One genuine failure remains: HNMD1V3, Seagate Z500R4KG, 2026-08-11.
257 lines
9.7 KiB
Python
257 lines
9.7 KiB
Python
"""
|
|
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 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()
|
|
|
|
|
|
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]
|