ge-aerospace/pxe-server
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

webapp: technician to-do for the manual CMMC steps; count fallbacks as wiped

Browse Source 
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.
This commit is contained in:
cproudlock
2026-10-05 13:50:26 -04:00
parent 44a7eeae95
commit 020f043d8b
5 changed files with 297 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files

84
webapp/services/blancco_evidence.py
View File

@@ -67,6 +67,7 @@ def record_cloud(document_id: str, sent: bool, serial: str = "", detail: str = "
"serial": (serial or "")[:64], "detail": (detail or "")[:200],
}
_save_state(state)
todo_invalidate()
def mark_servicenow(document_id: str, done: bool = True):
@@ -78,6 +79,7 @@ def mark_servicenow(document_id: str, done: bool = True):
else:
state["servicenow"].pop(document_id, None)
_save_state(state)
todo_invalidate()
def cloud_status(state: dict, document_id: str) -> str:
@@ -192,8 +194,11 @@ def servicenow_items(days: int = 30, include_done: bool = False) -> list[dict]:
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"]
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()
@@ -213,3 +218,78 @@ def servicenow_items(days: int = 30, include_done: bool = False) -> list[dict]:
"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

44
webapp/services/blancco_report.py
View File

@@ -167,6 +167,41 @@ def failure_reason(er: dict) -> str:
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.
@@ -191,7 +226,12 @@ def export_rows(filename: str, data: dict, when: str, cloud: str = "") -> list[d
"cloud_upload": cloud,
}
rows = []
for er in data.get("erasures") or []:
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 {}
@@ -211,6 +251,6 @@ def export_rows(filename: str, data: dict, when: str, cloud: str = "") -> list[d
"end_time": er.get("end_time", ""),
"elapsed_time": er.get("elapsed_time", ""),
"total_errors": er.get("total_errors", ""),
"failure_reason": failure_reason(er),
"failure_reason": reason,
})
return rows or [base]