"""CMMC evidence helpers for Blancco erasure reports. The onsite-support playbook asks techs for, among others: ENT-045 / 159 list of sanitized media + failure cause (ServiceNow Onsite Bin RITM "Wiped" / "Blancco Wipe failed" boxes) ENT-056 / 158 sample Blancco erasure certificates per site, monthly This module backs the webapp's evidence pack (one ZIP per site and month: summary.csv plus each report's XML and PDF), the ServiceNow tick list, and the record of which reports reached the Blancco cloud (posted by blancco-progress.py running inside Blancco). State lives in one small JSON file (config.BLANCCO_STATE), not on the report share, written atomically: {"cloud": {document_id: {"sent": bool, "ts", "serial", "detail"}}, "servicenow": {document_id: {"ts"}}} """ from __future__ import annotations import csv import io import json import os import re import tempfile import zipfile from datetime import datetime, timedelta, timezone import config from services import blancco_report _DOC_ID = re.compile(r"^[0-9a-fA-F-]{36}$") # --- state ------------------------------------------------------------------ def load_state() -> dict: try: with open(config.BLANCCO_STATE) as f: state = json.load(f) except (OSError, json.JSONDecodeError): state = {} state.setdefault("cloud", {}) state.setdefault("servicenow", {}) return state def _save_state(state: dict): folder = os.path.dirname(config.BLANCCO_STATE) os.makedirs(folder, exist_ok=True) fd, tmp = tempfile.mkstemp(dir=folder, prefix=".blancco-state-") with os.fdopen(fd, "w") as f: json.dump(state, f, indent=1, sort_keys=True) os.replace(tmp, config.BLANCCO_STATE) def _now() -> str: return datetime.now(timezone.utc).isoformat(timespec="seconds") def record_cloud(document_id: str, sent: bool, serial: str = "", detail: str = ""): if not _DOC_ID.match(document_id or ""): raise ValueError("bad document_id") state = load_state() state["cloud"][document_id] = { "sent": bool(sent), "ts": _now(), "serial": (serial or "")[:64], "detail": (detail or "")[:200], } _save_state(state) def mark_servicenow(document_id: str, done: bool = True): if not _DOC_ID.match(document_id or ""): raise ValueError("bad document_id") state = load_state() if done: state["servicenow"][document_id] = {"ts": _now()} else: state["servicenow"].pop(document_id, None) _save_state(state) def cloud_status(state: dict, document_id: str) -> str: """"sent", "failed", or "" when Blancco never told us.""" rec = state["cloud"].get(document_id) if not rec: return "" return "sent" if rec.get("sent") else "failed" # --- reports ---------------------------------------------------------------- def site_slug(site: str | None = None) -> str: return re.sub(r"[^A-Za-z0-9]+", "", site or config.BLANCCO_SITE) or "Site" def collect(since=None, until=None) -> list[dict]: """Every XML report on the share dated in [since, until), oldest first. Each entry: {"file", "pdf" (or None), "when", "data", "document_id"}. Dated by the timestamp inside the report, mtime only as a fallback. """ folder = config.BLANCCO_REPORTS names = set(os.listdir(folder)) if os.path.isdir(folder) else set() out = [] for f in sorted(names): path = os.path.join(folder, f) if not f.lower().endswith(".xml") or f.startswith(".") or not os.path.isfile(path): continue try: data = blancco_report.parse(path) except Exception: data = {} when = blancco_report.report_datetime(data) or datetime.fromtimestamp( os.stat(path).st_mtime, timezone.utc) if since and when < since: continue if until and when >= until: continue pdf = os.path.splitext(f)[0] + ".pdf" out.append({ "file": f, "pdf": pdf if pdf in names else None, "when": when, "data": data, "document_id": (data.get("meta") or {}).get("document_id", ""), }) return out def month_bounds(month: str): """'2026-10' -> (2026-10-01T00:00Z, 2026-11-01T00:00Z).""" y, m = (int(x) for x in month.split("-")) start = datetime(y, m, 1, tzinfo=timezone.utc) end = datetime(y + (m == 12), m % 12 + 1, 1, tzinfo=timezone.utc) return start, end def previous_month(now=None) -> str: now = now or datetime.now(timezone.utc) y, m = (now.year, now.month - 1) if now.month > 1 else (now.year - 1, 12) return f"{y:04d}-{m:02d}" def summary_csv(reports: list[dict], state: dict) -> str: buf = io.StringIO() writer = csv.DictWriter(buf, fieldnames=blancco_report.EXPORT_COLUMNS) writer.writeheader() for r in reports: writer.writerows(blancco_report.export_rows( r["file"], r["data"], r["when"].isoformat(), cloud_status(state, r["document_id"]))) return buf.getvalue() def evidence_name(month: str, site: str | None = None) -> str: return f"ENT-056_{site_slug(site)}_{month}" def build_evidence_zip(month: str, fileobj) -> int: """Write the evidence pack for `month` into fileobj; return report count.""" start, end = month_bounds(month) reports = collect(start, end) state = load_state() stem = evidence_name(month) with zipfile.ZipFile(fileobj, "w", zipfile.ZIP_DEFLATED) as zf: zf.writestr(f"{stem}/summary.csv", summary_csv(reports, state)) zf.writestr(f"{stem}/README.txt", ( f"Blancco erasure evidence, {config.BLANCCO_SITE}, {month} (UTC)\n" f"{len(reports)} report(s). summary.csv has one row per erased drive,\n" "with failure_reason for failed wipes and cloud_upload as recorded by\n" "the PXE server. Certificates of record are in the Blancco Management\n" "Portal; these are the site copies.\n")) for r in reports: zf.write(os.path.join(config.BLANCCO_REPORTS, r["file"]), f"{stem}/{r['file']}") if r["pdf"]: zf.write(os.path.join(config.BLANCCO_REPORTS, r["pdf"]), f"{stem}/{r['pdf']}") return len(reports) def servicenow_items(days: int = 30, include_done: bool = False) -> list[dict]: """Reports to reflect on the Onsite Bin Management RITM, newest first.""" since = datetime.now(timezone.utc) - timedelta(days=days) state = load_state() items = [] for r in reversed(collect(since)): done = state["servicenow"].get(r["document_id"]) if done and not include_done: continue data = r["data"] system = (data.get("hardware") or {}).get("system") or {} if isinstance(system, list): system = system[0] if system else {} erasures = data.get("erasures") or [] failed = [e for e in erasures if (e.get("state") or "").strip().lower() != "successful"] reasons = [blancco_report.failure_reason(e) for e in failed] drives = ", ".join( f"{(e.get('target') or {}).get('model', '')} {(e.get('target') or {}).get('serial', '')}".strip() for e in failed) items.append({ "document_id": r["document_id"], "file": r["file"], "when": r["when"], "serial": system.get("serial", ""), "model": system.get("model", ""), "failed": bool(failed) or not erasures, "note": (f"Blancco wipe failed on {drives or 'drive'}: " f"{'; '.join(x.rstrip('. ') for x in reasons if x) or 'no erasure recorded'}. " "Drive to be destroyed; record Certificate of Destruction.") if (failed or not erasures) else "", "cloud": cloud_status(state, r["document_id"]), "done": done.get("ts") if done else None, }) return items