diff --git a/webapp/app.py b/webapp/app.py index c570ab1..df362c2 100644 --- a/webapp/app.py +++ b/webapp/app.py @@ -565,17 +565,10 @@ def blancco_reports(): sysinfo = (data.get("hardware") or {}).get("system") or {} serial = sysinfo.get("serial", "") or "" model = sysinfo.get("model", "") or "" - # Overall erasure result: each erasure entry has its own - # 'state' (Successful / Failed / ...). If any drive failed - # the report rolls up to Failed; otherwise Successful. - erasures = data.get("erasures") or [] - states = [(e.get("state") or "").strip() for e in erasures if e.get("state")] - if not states: - result = "" - elif any(s.lower() != "successful" for s in states): - result = "Failed" - else: - result = "Successful" + # Overall result per drive: Failed if any drive was not + # wiped. A failed Purge followed by a successful Clear + # fallback on the same drive is a successful wipe. + result = blancco_report.report_result(data) cloud = blancco_evidence.cloud_status( state, (data.get("meta") or {}).get("document_id", "")) except Exception: @@ -724,6 +717,33 @@ def blancco_servicenow_mark(document_id): return redirect(url_for("blancco_servicenow", all=request.args.get("all"))) +@app.route("/reports/todo") +def blancco_todo(): + """Technician to-do: the playbook steps the server cannot do itself.""" + return render_template("reports_todo.html", todo=blancco_evidence.todo(), + site=config.BLANCCO_SITE) + + +@app.route("/reports/todo//", methods=["POST"]) +def blancco_todo_mark(kind, key): + done = request.form.get("done", "1") == "1" + try: + blancco_evidence.mark(kind, key, done) + except ValueError: + abort(400) + audit("REPORT_TODO", f"{kind} {key} {'done' if done else 'undone'}") + return redirect(url_for("blancco_todo")) + + +@app.route("/reports/evidence/file/") +def blancco_evidence_file(filename): + filename = secure_filename(filename) + fpath = os.path.join(config.BLANCCO_REPORTS, "evidence", filename) + if not os.path.isfile(fpath): + abort(404) + return send_file(fpath, as_attachment=True) + + @app.route("/reports/download/") def blancco_download_report(filename): filename = secure_filename(filename) @@ -1481,6 +1501,7 @@ def inject_globals(): return { "all_image_types": config.IMAGE_TYPES, "all_friendly_names": config.FRIENDLY_NAMES, + "blancco_todo": blancco_evidence.todo_counts(), } diff --git a/webapp/services/blancco_evidence.py b/webapp/services/blancco_evidence.py index 4d2f73a..677f1d6 100644 --- a/webapp/services/blancco_evidence.py +++ b/webapp/services/blancco_evidence.py @@ -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 diff --git a/webapp/services/blancco_report.py b/webapp/services/blancco_report.py index 0621eb6..cea2660 100644 --- a/webapp/services/blancco_report.py +++ b/webapp/services/blancco_report.py @@ -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] diff --git a/webapp/templates/base.html b/webapp/templates/base.html index f5dc74e..abc2c90 100644 --- a/webapp/templates/base.html +++ b/webapp/templates/base.html @@ -75,6 +75,13 @@ Blancco Reports +