#!/usr/bin/env python3 """Report Blancco erase progress to the PXE webapp's /imaging dashboard. Runs inside the Blancco rootfs (installed by blancco-init.sh as a systemd service). Follows the AlbusSeven journal and maps its log lines onto the same /imaging/status payload the WinPE imaging clients post, so each wiping PC gets a tile keyed by BIOS serial with pctype "blancco". Stages (stage_total 7): 1 booting and 2 starting are posted by blancco-init.sh; this script posts 3 waiting for licence, 4 erasing, 5 report saved, 6 report sent / on share, 7 done. A failed erasure step marks the session failed. The final state comes from the report itself, not the journal: outside Blancco's debug mode Albus does not log "erasureFinishedForEntity", "Report saving success" or "Auto process reached the final step", so a watcher on /reports (where auto mode always saves the report) reads each new report and posts 7 "Wipe complete" or "Erasure FAILED". Usage: blancco-progress.py [--replay FILE] [--dry-run] """ import glob import json import os import re import subprocess import sys import threading import time import urllib.request import xml.etree.ElementTree as ET STATUS_URL = "http://172.16.9.1:9009/imaging/status" CLOUD_URL = "http://172.16.9.1:9009/reports/cloud-status" STAGE_TOTAL = 7 # Progress lines arrive in bursts; post at most this often unless the # stage itself changes. MIN_POST_INTERVAL = 3.0 def read(path): try: with open(path) as f: return f.read().strip() except OSError: return "" def pxe_iface(): """(mac, ip) of the interface on the PXE LAN, or blanks.""" try: out = subprocess.run(["ip", "-o", "-4", "addr"], capture_output=True, text=True, timeout=5).stdout except (OSError, subprocess.SubprocessError): return "", "" for line in out.splitlines(): m = re.search(r"^\d+:\s+(\S+)\s+inet (172\.16\.9\.\d+)/", line) if m: return read(f"/sys/class/net/{m.group(1)}/address"), m.group(2) return "", "" REPORTS_DIR = "/reports" def report_outcome(path): """(wiped: bool, failed_drives: list[str]) from a Blancco report XML. Per drive: wiped if any of its erasures succeeded, so a failed NIST Purge followed by a successful Clear fallback still counts as wiped. None if the report has no erasures (hardware-only report). """ root = ET.parse(path).getroot() drives = {} for i, er in enumerate(root.iter("entries")): if er.get("name") != "erasure": continue state, serial, model = "", "", "" for e in er: if e.tag == "entry" and e.get("name") == "state": state = (e.text or "").strip() if e.tag == "entries" and e.get("name") == "target": for t in e: if t.get("name") == "serial": serial = (t.text or "").strip() elif t.get("name") == "model": model = (t.text or "").strip() key = serial or f"#{i}" ok = state.lower() == "successful" prev = drives.get(key, (False, "")) drives[key] = (prev[0] or ok, f"{model} {serial}".strip()) if not drives: return None failed = [label for ok, label in drives.values() if not ok] return (not failed, failed) SERIAL = read("/sys/class/dmi/id/product_serial") or read("/etc/machine-id") MODEL = " ".join(x for x in (read("/sys/class/dmi/id/sys_vendor"), read("/sys/class/dmi/id/product_name")) if x) # (pattern, stage_index, stage text or callable(match) -> text, status) RULES = [ (re.compile(r"Waiting for MC connectivity"), 3, "Waiting for licence (cloud)", None), (re.compile(r"ClurichaunLicenseProvider: ([1-9]\d*) licenses available for product id 1\b"), 3, lambda m: f"Licence OK ({m.group(1)} left)", None), (re.compile(r"startErasure: success"), 4, "Erasing", None), (re.compile(r"About to start Step (\d+) \((.*?)\)\.?$"), 4, lambda m: f"Erasing: step {m.group(1)} - {m.group(2).split(' or ')[0][:60]}", None), (re.compile(r"Step \d+ \(.*\) failed|Erasure step .* failed\b|erasure failed", re.I), 4, "Erasure FAILED", "failed"), (re.compile(r"erasureFinishedForEntity"), 5, "Erasure finished", None), (re.compile(r"\[report\.manager\].*Report saving success"), 5, "Report saved", None), (re.compile(r"Report sent to MC"), 6, "Report sent to cloud", None), (re.compile(r"Auto process reached the final step"), 7, "Done", "succeeded"), ] # Cloud upload of a report: "[comm.manager] Sending report document id: " # then "Report sent to MC" (or an upload failure). SENDING = re.compile(r"Sending report document id: ([0-9a-fA-F-]{36})") SENT = re.compile(r"Report sent to MC") SEND_FAILED = re.compile(r"(report (upload|send|sending)[^\n]*fail|failed to (send|upload) report)", re.I) # Drive progress, e.g. "FORMAT NVM progress: 75%". Excludes the self-test # completion and volume lines that also carry a percentage. PROGRESS = re.compile(r"\bprogress[^%\d]{0,20}(\d{1,3})%", re.I) class Reporter: def __init__(self, dry_run=False): self.dry_run = dry_run self.mac, self.ip = pxe_iface() self.stage_index = 2 self.stage = "Blancco started" self.status = "in_progress" self.last_post = 0.0 self.pending = [] def post(self, force=False): now = time.monotonic() if not force and now - self.last_post < MIN_POST_INTERVAL: return payload = { "serial": SERIAL, "mac": self.mac, "ip": self.ip, "pctype": "blancco", "hostname_target": MODEL, "current_stage": self.stage, "stage_index": self.stage_index, "stage_total": STAGE_TOTAL, "status": self.status, "source": "blancco", } if self.pending: payload["log_lines"] = self.pending[-20:] if self.dry_run: print(json.dumps(payload)) else: req = urllib.request.Request( STATUS_URL, data=json.dumps(payload).encode(), headers={"Content-Type": "application/json"}) try: urllib.request.urlopen(req, timeout=5).read() except OSError: return # webapp unreachable: keep the lines for the next post self.pending = [] self.last_post = now def cloud(self, document_id, sent, detail): """Tell the webapp whether a report reached the Blancco cloud.""" payload = {"document_id": document_id, "serial": SERIAL, "sent": sent, "detail": detail[:200]} if self.dry_run: print(json.dumps({"cloud": payload})) return req = urllib.request.Request( CLOUD_URL, data=json.dumps(payload).encode(), headers={"Content-Type": "application/json"}) try: urllib.request.urlopen(req, timeout=5).read() except OSError: pass def line(self, text): text = text.strip() if not text: return m = SENDING.search(text) if m: self.sending = m.group(1) elif getattr(self, "sending", None): if SENT.search(text): self.cloud(self.sending, True, text) self.sending = None elif SEND_FAILED.search(text): self.cloud(self.sending, False, text) self.sending = None for pat, idx, stage, status in RULES: m = pat.search(text) if not m: continue if self.status == "failed" and status != "failed": return # a failure sticks for the rest of the run if idx < self.stage_index and status != "failed": return # late licence chatter must not rewind the tile new_stage = stage(m) if callable(stage) else stage changed = (idx, new_stage) != (self.stage_index, self.stage) self.stage_index = max(self.stage_index, idx) if status != "failed" else idx self.stage = new_stage if status: self.status = status self.pending.append(text[:200]) if changed: self.post(force=True) return m = PROGRESS.search(text) if m and self.stage_index == 4 and "Self-Test" not in text: base = self.stage.split(" [")[0] self.stage = f"{base} [{m.group(1)}%]" self.post() def finish(self, path): """Final stage from a finished report on disk.""" try: outcome = report_outcome(path) except (OSError, ET.ParseError): return False if outcome is None: return True # nothing erased in this report; keep watching wiped, failed = outcome name = os.path.basename(path) self.stage_index = 7 if wiped: self.stage, self.status = "Wipe complete", "succeeded" else: self.stage, self.status = "Erasure FAILED: " + ", ".join(failed)[:120], "failed" self.pending.append(f"Report {name}: {'all drives wiped' if wiped else 'failed: ' + ', '.join(failed)}") self.post(force=True) return True def watch_reports(rep, interval=5, stable=5): """Post the final stage for each new report Albus saves in /reports.""" seen = set(glob.glob(os.path.join(REPORTS_DIR, "*.xml"))) while True: time.sleep(interval) for path in sorted(glob.glob(os.path.join(REPORTS_DIR, "*.xml"))): if path in seen: continue try: st = os.stat(path) except OSError: continue if st.st_size == 0 or time.time() - st.st_mtime < stable: continue # Albus may still be writing it if rep.finish(path): seen.add(path) def main(): dry_run = "--dry-run" in sys.argv rep = Reporter(dry_run=dry_run) if "--outcome" in sys.argv: print(report_outcome(sys.argv[sys.argv.index("--outcome") + 1])) return if "--replay" in sys.argv: with open(sys.argv[sys.argv.index("--replay") + 1]) as f: for text in f: rep.line(text) rep.post(force=True) return rep.post(force=True) threading.Thread(target=watch_reports, args=(rep,), daemon=True).start() proc = subprocess.Popen( ["journalctl", "-f", "-b", "-o", "cat", "-t", "AlbusSeven", "-n", "all"], stdout=subprocess.PIPE, text=True, bufsize=1) for text in proc.stdout: rep.line(text) if rep.status in ("succeeded", "failed") and rep.stage_index == STAGE_TOTAL: rep.post(force=True) if __name__ == "__main__": main()