pxe-server/webapp/services/blancco_evidence.py

"""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}$")
# Reports without a usable document id (older or hand-made ones) are tracked
# by file name instead, so they can still be ticked and cleared.
_FILE_KEY = re.compile(r"^file:[A-Za-z0-9._-]{1,200}$")
_REPORT_KEY = re.compile(f"{_DOC_ID.pattern}|{_FILE_KEY.pattern}")


def report_key(document_id: str, filename: str) -> str:
    """The state-file key for one report: its document id, else "file:<name>"."""
    return document_id if _DOC_ID.match(document_id or "") else f"file:{filename}"


# --- 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)
    todo_invalidate()


def mark_servicenow(document_id: str, done: bool = True):
    if not _REPORT_KEY.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)
    todo_invalidate()


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", ""),
        })
        out[-1]["key"] = report_key(out[-1]["document_id"], f)
    return out


def _tz():
    from zoneinfo import ZoneInfo
    return ZoneInfo(config.DISPLAY_TZ)


def month_bounds(month: str):
    """'2026-10' -> start of that month and of the next, in the site's
    display timezone (a wipe at 21:00 EDT on the 31st belongs to that
    month, not the next one as it would in UTC)."""
    y, m = (int(x) for x in month.split("-"))
    start = datetime(y, m, 1, tzinfo=_tz())
    end = datetime(y + (m == 12), m % 12 + 1, 1, tzinfo=_tz())
    return start, end


def previous_month(now=None) -> str:
    now = now or datetime.now(_tz())
    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"].astimezone(_tz()).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} ({config.DISPLAY_TZ})\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 failure_note(data: dict):
    """(failed erasures, ServiceNow note) for a parsed report. A drive is
    failed only if none of its erasures succeeded (a Purge failure followed
    by a successful Clear fallback is a good wipe). The note is "" when every
    drive was wiped, and also covers a report with no erasure at all."""
    erasures = data.get("erasures") or []
    outcomes = blancco_report.drive_outcomes(data)
    failed = [e for i, e in enumerate(erasures)
              if not outcomes[blancco_report._drive_key(e, i)]["ok"]]
    if erasures and not failed:
        return failed, ""
    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)
    return failed, (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.")


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["key"])
        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, note = failure_note(data)
        items.append({
            "document_id": r["document_id"],
            "key": r["key"],
            "file": r["file"],
            "when": r["when"].astimezone(_tz()),
            "serial": system.get("serial", ""),
            "model": system.get("model", ""),
            "failed": bool(failed) or not erasures,
            "note": note,
            "cloud": cloud_status(state, r["document_id"]),
            "done": done.get("ts") if done else None,
        })
    return items


# --- technician to-do -------------------------------------------------------
# The parts of the playbook the server cannot do itself. Each kind is a dict
# in the state file keyed by month or document_id.
TODO_KINDS = {
    "evidence_uploaded": re.compile(r"^20\d\d-(0[1-9]|1[0-2])$"),  # month
    "destroyed": _REPORT_KEY,       # failed wipe: Certificate of Destruction recorded
    "cloud_resolved": _REPORT_KEY,  # report not in the cloud: uploaded by hand
}
_EVIDENCE_ZIP = re.compile(r"^ENT-056_.+_(20\d\d-\d\d)\.zip$")


def mark(kind: str, key: str, done: bool = True):
    pattern = TODO_KINDS.get(kind)
    if not pattern or not pattern.match(key or ""):
        raise ValueError("bad to-do item")
    state = load_state()
    bucket = state.setdefault(kind, {})
    if done:
        bucket[key] = {"ts": _now()}
    else:
        bucket.pop(key, None)
    _save_state(state)
    todo_invalidate()


def todo() -> dict:
    """Open technician tasks, newest first in each list."""
    state = load_state()
    for kind in TODO_KINDS:
        state.setdefault(kind, {})
    evidence_dir = os.path.join(config.BLANCCO_REPORTS, "evidence")
    packs = []
    if os.path.isdir(evidence_dir):
        for f in sorted(os.listdir(evidence_dir), reverse=True):
            m = _EVIDENCE_ZIP.match(f)
            if m and m.group(1) not in state["evidence_uploaded"]:
                packs.append({"month": m.group(1), "file": f})
    year = servicenow_items(days=366, include_done=True)
    failed = [i for i in year if i["failed"] and i["key"] not in state["destroyed"]]
    not_in_cloud = [i for i in year
                    if i["cloud"] == "failed" and i["key"] not in state["cloud_resolved"]]
    servicenow = sum(1 for i in year if not i["done"]
                     and i["when"] >= datetime.now(timezone.utc) - timedelta(days=30))
    return {
        "servicenow": servicenow,
        "evidence": packs,
        "failed": failed,
        "not_in_cloud": not_in_cloud,
        "total": servicenow + len(packs) + len(failed) + len(not_in_cloud),
    }


_todo_cache = {"at": 0.0, "value": None}


def todo_counts(ttl: float = 60.0) -> dict:
    """todo() reduced to counts, cached: it is shown on every page."""
    import time
    now = time.monotonic()
    if _todo_cache["value"] is None or now - _todo_cache["at"] > ttl:
        try:
            t = todo()
            value = {"servicenow": t["servicenow"], "evidence": len(t["evidence"]),
                     "failed": len(t["failed"]), "not_in_cloud": len(t["not_in_cloud"]),
                     "total": t["total"]}
        except Exception:
            value = {"servicenow": 0, "evidence": 0, "failed": 0, "not_in_cloud": 0, "total": 0}
        _todo_cache.update(at=now, value=value)
    return _todo_cache["value"]


def todo_invalidate():
    _todo_cache["value"] = None