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.
296 lines
11 KiB
Python
296 lines
11 KiB
Python
"""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)
|
|
todo_invalidate()
|
|
|
|
|
|
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)
|
|
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", ""),
|
|
})
|
|
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 []
|
|
outcomes = blancco_report.drive_outcomes(data)
|
|
# A drive is failed only if none of its erasures succeeded (a Purge
|
|
# failure followed by a successful Clear fallback is a good wipe).
|
|
failed = [e for i, e in enumerate(erasures)
|
|
if not outcomes[blancco_report._drive_key(e, i)]["ok"]]
|
|
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
|
|
|
|
|
|
# --- 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": _DOC_ID, # failed wipe: Certificate of Destruction recorded
|
|
"cloud_resolved": _DOC_ID, # 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["document_id"] not in state["destroyed"]]
|
|
not_in_cloud = [i for i in year
|
|
if i["cloud"] == "failed" and i["document_id"] 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
|