#!/usr/bin/env python3 """Flask web application for managing a GE Aerospace PXE server. This file is the route surface; most logic lives in ``services/``: services.audit - audit log writer services.csrf - session CSRF token + before_request validator services.fs - path helpers + JSON load/save services.system - systemd service status + USB mounts services.images - image_status + load_image_config services.deploy - import_deploy + merge_tree + symlink dance services.unattend - parse + build + form-extract for unattend.xml services.wim - boot.wim startnet.cmd extract/update via wimtools services.image_pages - per-image page tabs, header + overview data services.control_files - preview/backup/atomic write/undo of control files The Updates pages are blueprints in ``views/`` (registered below). """ import hashlib import json import os import re import shutil import tempfile import time from datetime import datetime, timedelta, timezone from pathlib import Path from flask import ( Flask, Response, abort, flash, jsonify, redirect, render_template, request, send_file, stream_with_context, url_for, ) from werkzeug.utils import secure_filename import config from services import ( blancco_evidence, clonezilla_images, hardware_coverage, blancco_report, dell_updates, deploy, driver_map, fs, control_files, image_pages, image_registry, images, imaging_status, mcl_ppkg, nic_status, system, unattend, wim, ) from services.audit import audit from services.csrf import init_csrf from views import blancco_update as blancco_update_views from views import bprt_sync as bprt_sync_views from views import dell_updates as dell_updates_views from views import images as images_views from views import images_drivers as images_drivers_views from views import images_unattend as images_unattend_views from views import ipxe as ipxe_views app = Flask(__name__) app.secret_key = config.FLASK_SECRET_KEY app.config["MAX_CONTENT_LENGTH"] = config.MAX_CONTENT_LENGTH init_csrf(app) # Pull IMAGE_TYPES/FRIENDLY_NAMES from the registry file (created from the # baked-in config.py defaults on first run). Mutates config.* in place. image_registry.reload() # Updates pages (Dell drivers/BIOS, iPXE, Blancco, BPRT sync): one blueprint # each in views/, logic in services/; all routes live under /updates/. app.register_blueprint(dell_updates_views.bp) app.register_blueprint(ipxe_views.bp) app.register_blueprint(blancco_update_views.bp) app.register_blueprint(bprt_sync_views.bp) # GE Aerospace image pages: one page per image with tabs (Overview, Drivers, # Unattend, OS and Packages) under templates/_image_header.html. The tab -> # endpoint table is services/image_pages.IMAGE_TABS; unattend_editor below # is the Unattend tab, the others are blueprints. app.register_blueprint(images_views.bp) app.register_blueprint(images_drivers_views.bp) app.register_blueprint(images_unattend_views.bp) # Unattend review/save/undo # --------------------------------------------------------------------------- # Routes - pages # --------------------------------------------------------------------------- def _nic_parts(): """(pill, notice) HTML for the Overview's network status. Rendered here, each inside a try, so a bad status file or a template slip shows 'Network: unknown' and never takes the Overview down.""" from markupsafe import Markup try: nic = nic_status.status() except Exception as ex: # the page must render even if this breaks app.logger.exception("network status failed") nic = nic_status.unknown(ex) parts = [] for part in ("pill", "notice"): try: parts.append(Markup(render_template("_nic_status.html", nic=nic, part=part))) except Exception as ex: app.logger.exception("network status render failed") parts.append(Markup(render_template("_nic_status.html", nic=nic_status.unknown(ex), part=part)) if part == "pill" else Markup("")) return parts[0], parts[1] @app.route("/") def dashboard(): image_list = [images.image_status(it) for it in config.IMAGE_TYPES] for img in image_list: # Card details: OS, driver entries and lint, from the image pages. try: info = image_pages.header_info(img["image_type"]) img["os_label"] = info["os_label"] img["driver_count"] = info["driver_count"] except Exception: img["os_label"], img["driver_count"] = "", 0 img["lint"] = image_pages._lint_counts(img["image_type"]) services = [system.service_status(s) for s in ("dnsmasq", "apache2", "smbd")] nic_pill, nic_notice = _nic_parts() sessions, _hidden = imaging_status.hide_stale(imaging_status.list_sessions()) floor = imaging_status.floor_order(sessions) real = [x for x in floor if x["floor_state"] != "logs"] counts = {k: sum(1 for x in real if x["floor_state"] == k) for k in ("failed", "action", "running", "done")} return render_template( "dashboard.html", images=image_list, services=services, nic_pill=nic_pill, nic_notice=nic_notice, image_types=config.IMAGE_TYPES, friendly_names=config.FRIENDLY_NAMES, sessions=real[:6], more=max(0, len(real) - 6), floor_counts=counts, compact=True, ) @app.route("/updates/") def updates_overview(): """One card per updater (Dell, iPXE, Blancco, BPRT) and the Dell applies with their rollback state. Each card links to its own page.""" from services import blancco_update, bprt_sync, dell_updates, ipxe_update, updates_common try: scan = dell_updates.last_scan() except Exception: # unreadable scan.json: show "not checked" rather than a 500 scan = None dell = {"checked": scan.get("checked") if scan else None, "drivers": len(scan.get("drivers") or []) if scan else 0, "bios": len(scan.get("bios") or []) if scan else 0, "drv_updates": sum(1 for r in (scan.get("drivers") or []) if r.get("status") == "update") if scan else 0, "bios_updates": sum(1 for r in (scan.get("bios") or []) if r.get("status") == "update") if scan else 0} try: rows = ipxe_update.installed() by_name = {r["name"]: r["info"] for r in rows} latest = ipxe_update.last_check() efi = by_name.get("ipxe.efi") or {} ipxe = {"rows": rows, "version": efi.get("version"), "latest": latest, "status": ipxe_update.compare(latest["tag"], efi.get("version")) if latest else None} except Exception as ex: ipxe = {"rows": [], "version": None, "latest": None, "status": None, "error": str(ex)} try: blancco = {"current": blancco_update.current(), "staged": blancco_update.staged()} except Exception as ex: # a broken release folder must not take the page down blancco = {"current": {}, "staged": [], "error": str(ex)} try: bprt = {"live": bprt_sync.live_summary(), "batches": bprt_sync.batches()} except Exception as ex: bprt = {"live": {}, "batches": [], "error": str(ex)} return render_template("updates_overview.html", dell=dell, ipxe=ipxe, blancco=blancco, bprt=bprt, history=dell_updates.history_view(10), settings=updates_common.settings()) @app.route("/images/import", methods=["GET", "POST"]) def images_import(): usb_mounts = system.find_usb_mounts() upload_sources = system.find_upload_sources() image_list = [images.image_status(it) for it in config.IMAGE_TYPES] if request.method == "POST": source = request.form.get("source", "") target = request.form.get("target", "") if not source or not target: flash("Please select both a source and a target image type.", "danger") return redirect(url_for("images_import")) if target not in config.IMAGE_TYPES: flash("Invalid target image type.", "danger") return redirect(url_for("images_import")) if not deploy.allowed_import_source(source): flash("Source path is not a valid import location.", "danger") return redirect(url_for("images_import")) if not os.path.isdir(source): flash(f"Source path does not exist: {source}", "danger") return redirect(url_for("images_import")) root = fs.image_root(target) dest = fs.deploy_path(target) try: os.makedirs(dest, exist_ok=True) src_items = os.listdir(source) # Move files from network upload to save disk space; copy from USB. # deploy.sync_tree uses rsync --checksum: unchanged files are # skipped, only new/changed files are written (no full replace). use_move = source == config.UPLOAD_DIR or source.startswith(config.UPLOAD_DIR + "/") top_dirs = {d for d in src_items if os.path.isdir(os.path.join(source, d))} full_layout = "Deploy" in top_dirs if full_layout: shared_root = [] for prefix, dirs in config.SHARED_ROOT_DIRS.items(): if target.startswith(prefix): shared_root = dirs break for item in src_items: src_item = os.path.join(source, item) if item == "Deploy": deploy.import_deploy(src_item, dest, target, move=use_move) elif os.path.isdir(src_item) and item in shared_root: prefix_key = target.split("-")[0] + "-" shared_dest = os.path.join(config.SHARED_DIR, f"{prefix_key}{item}") deploy.sync_tree(src_item, shared_dest, move=use_move) dst_item = os.path.join(root, item) if os.path.islink(dst_item): os.remove(dst_item) elif os.path.isdir(dst_item): shutil.rmtree(dst_item) os.symlink(shared_dest, dst_item) elif os.path.isdir(src_item): deploy.sync_tree(src_item, os.path.join(root, item), move=use_move) else: deploy.sync_tree(src_item, os.path.join(root, item), move=use_move) else: deploy.import_deploy(source, dest, target, move=use_move) # Ensure Media.tag exists (FlatSetupLoader.exe drive detection). control_dir = os.path.join(dest, "Control") os.makedirs(control_dir, exist_ok=True) media_tag = os.path.join(control_dir, "Media.tag") Path(media_tag).touch() audit("IMAGE_IMPORT", f"{source} -> {target}") flash( f"Successfully imported content to {config.FRIENDLY_NAMES.get(target, target)}.", "success", ) except Exception as exc: flash(f"Import failed: {exc}", "danger") return redirect(url_for("images_import")) return render_template( "import.html", usb_mounts=usb_mounts, upload_sources=upload_sources, images=image_list, image_types=config.IMAGE_TYPES, friendly_names=config.FRIENDLY_NAMES, ) @app.route("/images//unattend", methods=["GET", "POST"]) def unattend_editor(image_type): if image_type not in config.IMAGE_TYPES: flash("Unknown image type.", "danger") return redirect(url_for("dashboard")) xml_file = fs.unattend_path(image_type) if request.method == "POST": # Plain form post (the page itself saves through Review changes, # views/images_unattend.py). Same path: the form is applied to the # current text, backup, undo point, lint gate. save_mode = request.form.get("save_mode", "form") if save_mode == "raw": body = {"mode": "raw", "raw_xml": request.form.get("raw_xml", "")} else: body = {"mode": "form", "form": unattend.extract_form_data(request.form)} images_unattend_views.save_and_flash(image_type, body, f"Unattend save ({save_mode}, form post)") return redirect(url_for("unattend_editor", image_type=image_type)) data = unattend.parse_unattend(xml_file) return render_template( "unattend_editor.html", image_type=image_type, friendly_name=config.FRIENDLY_NAMES.get(image_type, image_type), active_tab="unattend", data=data, image_types=config.IMAGE_TYPES, friendly_names=config.FRIENDLY_NAMES, **images_unattend_views.page_context(image_type), ) @app.route("/images//config") def image_config(image_type): """Old single Configuration page, now split into tabs. Bookmarks land on the Overview, or on one tab with ?tab=drivers|os|unattend.""" if image_type not in config.IMAGE_TYPES: flash("Unknown image type.", "danger") return redirect(url_for("dashboard")) tab = request.args.get("tab", "overview") if tab not in image_pages.TAB_IDS: tab = "overview" return redirect(image_pages.tab_url(image_type, tab)) # Legacy /config/save sections and the tab each one now lives on. _CONFIG_SECTIONS = { "hardware_models": "os", "drivers": "drivers", "operating_systems": "os", "packages": "os", } @app.route("/images//config/save", methods=["POST"]) def image_config_save(image_type): """Old one-section form save. No page posts here any more, and it must not write: it would skip the diff preview, the stale-file check and the Drivers tab's lint gate. It sends the user to the tab that saves now.""" if image_type not in config.IMAGE_TYPES: flash("Unknown image type.", "danger") return redirect(url_for("dashboard")) tab = _CONFIG_SECTIONS.get(request.form.get("section", ""), "overview") flash("This form no longer saves. Make the change on this tab and use Review and save; " "nothing was written.", "warning") return redirect(image_pages.tab_url(image_type, tab)) def _register_package(image_type, fname, dest): """Append {fileName, destinationDir} to packages.json through the same path as the OS and Packages tab (format kept, .bak--webui backup, atomic, the tab's undo point). False when it is already listed.""" path = image_pages.control_file(image_type, "packages.json") text = control_files.read_text(path) entries = [] if text.strip(): try: entries = json.loads(text.lstrip("\ufeff")) except ValueError: entries = None if not isinstance(entries, list): raise ValueError("packages.json is not a JSON list, so the file was not registered") if any(str(e.get("fileName") or e.get("FileName") or "").lower() == fname.lower() for e in entries if isinstance(e, dict)): return False entries.append({"fileName": fname, "destinationDir": dest}) plan = control_files.plan_json(path, entries, "packages.json") control_files.commit(image_type, images_views.UNDO_SCOPE, [plan], action=f"Package upload: {fname} registered") return True # --------------------------------------------------------------------------- # Routes - Driver/package upload + orphan adoption # --------------------------------------------------------------------------- @app.route("/images//drivers/upload", methods=["POST"]) def images_drivers_upload(image_type): """Store a driver .zip under Out-of-box Drivers (services/driver_map. save_upload: temp file + rename, in-use zips never overwritten). It is NOT added to HardwareDriver.json here: the Drivers tab's Add to map does that through the review step, with a backup and Undo. family and destination_dir from old forms are ignored for that reason.""" if image_type not in config.IMAGE_TYPES: flash(f"Unknown image type: {image_type}", "danger") return redirect(url_for("dashboard")) back = redirect(image_pages.tab_url(image_type, "drivers")) f = request.files.get("driver_file") if not f or not f.filename: flash("No file selected.", "danger") return back overwrite = request.form.get("overwrite") == "1" try: rec = driver_map.save_upload(image_type, f, overwrite=overwrite) audit("DRIVER_UPLOAD", f"{image_type}/{rec['filename']} replaced={rec['replaced']}") flash(f"Uploaded {rec['filename']}. It is listed under Unregistered zips; " "use Add to map there to give it an entry.", "success") except (ValueError, FileExistsError) as ex: flash(str(ex), "danger") except Exception as ex: flash(f"Upload failed: {ex}", "danger") return back @app.route("/images//drivers/adopt", methods=["POST"]) def images_drivers_adopt(image_type): """Old one-click adopt. It rewrote HardwareDriver.json with indent 2, no backup and an entry with no modelswminame (PESetup throws on that), so it now writes nothing and points at Add to map on the Drivers tab.""" if image_type not in config.IMAGE_TYPES: flash(f"Unknown image type: {image_type}", "danger") return redirect(url_for("dashboard")) flash("Use Add to map under Unregistered zips: the new entry needs model tokens and is " "saved through Review and save. Nothing was written.", "info") return redirect(image_pages.tab_url(image_type, "drivers")) @app.route("/images//drivers/orphans/delete", methods=["POST"]) def images_drivers_orphans_delete(image_type): """Old orphan delete (form field "filename"), now the same safe delete as the Drivers tab: zips any image, undo or rollback uses are kept.""" if image_type not in config.IMAGE_TYPES: flash(f"Unknown image type: {image_type}", "danger") return redirect(url_for("dashboard")) back = redirect(image_pages.tab_url(image_type, "drivers")) filenames = request.form.getlist("filename") if not filenames: flash("No files selected for removal.", "warning") return back rec = driver_map.delete_orphans(image_type, filenames) audit("DRIVER_ORPHAN_REMOVE", f"{image_type} removed={len(rec['removed'])} " f"refused={len(rec['refused'])} {', '.join(rec['removed'])}") if rec["removed"]: flash(f"Deleted {len(rec['removed'])} unregistered zip(s).", "success") for rel, why in rec["refused"]: flash(f"Not deleted: {rel} ({why}).", "warning") return back @app.route("/images//packages/upload", methods=["POST"]) def images_packages_upload(image_type): if image_type not in config.IMAGE_TYPES: flash(f"Unknown image type: {image_type}", "danger") return redirect(url_for("dashboard")) f = request.files.get("package_file") if not f or not f.filename: flash("No file selected.", "danger") return redirect(image_pages.tab_url(image_type, "os")) dest = (request.form.get("destination_dir") or "").strip() overwrite = request.form.get("overwrite") == "1" try: # The file only; packages.json is edited here through control_files # (services.images.upload_package would rewrite it without a backup). rec = images.upload_package(image_type, f, destination_dir="", overwrite=overwrite) msg = f"Uploaded {rec['filename']}." registered = False if dest: try: registered = _register_package(image_type, rec["filename"], dest) msg += (" Registered in packages.json (backup made; Undo last save puts it back)." if registered else " Already listed in packages.json.") except (ValueError, OSError) as ex: flash(f"Not registered: {ex}", "warning") else: msg += " Not registered (no destination_dir provided)." audit("PACKAGE_UPLOAD", f"{image_type}/{rec['filename']} registered={registered}") flash(msg, "success") except (ValueError, FileExistsError) as ex: flash(str(ex), "danger") except Exception as ex: flash(f"Upload failed: {ex}", "danger") return redirect(image_pages.tab_url(image_type, "os")) # --------------------------------------------------------------------------- # Routes - Image type CRUD (registry-backed) # --------------------------------------------------------------------------- @app.route("/images/new", methods=["POST"]) def images_new(): key = (request.form.get("key") or "").strip().lower() friendly_name = (request.form.get("friendly_name") or "").strip() try: rec = image_registry.create(key, friendly_name) audit("IMAGE_REGISTRY_CREATE", f"{rec['key']} ({rec['friendly_name']})") flash(f"Created image type {rec['key']}.", "success") except image_registry.RegistryError as ex: flash(str(ex), "danger") return redirect(url_for("dashboard")) @app.route("/images//clone", methods=["POST"]) def images_clone(image_type): if image_type not in config.IMAGE_TYPES: flash(f"Unknown source image type: {image_type}", "danger") return redirect(url_for("dashboard")) dst_key = (request.form.get("dst_key") or "").strip().lower() friendly_name = (request.form.get("friendly_name") or "").strip() or None try: rec = image_registry.clone(image_type, dst_key, friendly_name) audit("IMAGE_REGISTRY_CLONE", f"{image_type} -> {rec['key']}") flash(f"Cloned {image_type} -> {rec['key']}.", "success") except image_registry.RegistryError as ex: flash(str(ex), "danger") except Exception as ex: flash(f"Clone failed: {ex}", "danger") return redirect(url_for("dashboard")) @app.route("/images//delete", methods=["POST"]) def images_delete(image_type): if image_type not in config.IMAGE_TYPES: flash(f"Unknown image type: {image_type}", "danger") return redirect(url_for("dashboard")) delete_content = request.form.get("delete_content") == "1" try: rec = image_registry.delete(image_type, delete_content=delete_content) audit("IMAGE_REGISTRY_DELETE", f"{rec['key']} content={rec['removed_content']}") msg = f"Removed image type {rec['key']} from registry." if rec["removed_content"]: msg += " On-disk content wiped." flash(msg, "success") except image_registry.RegistryError as ex: flash(str(ex), "danger") return redirect(url_for("dashboard")) @app.route("/images//rename", methods=["POST"]) def images_rename(image_type): if image_type not in config.IMAGE_TYPES: flash(f"Unknown image type: {image_type}", "danger") return redirect(url_for("dashboard")) friendly_name = (request.form.get("friendly_name") or "").strip() try: rec = image_registry.rename_friendly(image_type, friendly_name) audit("IMAGE_REGISTRY_RENAME", f"{rec['key']} -> {rec['friendly_name']}") flash(f"Renamed {rec['key']} to '{rec['friendly_name']}'.", "success") except image_registry.RegistryError as ex: flash(str(ex), "danger") return redirect(url_for("dashboard")) # --------------------------------------------------------------------------- # Routes - Clonezilla Backups # --------------------------------------------------------------------------- def _backups_page(plan=None, plan_keep=None): images = clonezilla_images.list_images() zips = clonezilla_images.list_zips() groups = clonezilla_images.grouped(images, zips) st = clonezilla_images.settings() return render_template( "backups.html", groups=groups, backups=zips, image_count=len(images), mismatches=sum(1 for g in groups if g["mismatch"]), space=clonezilla_images.disk_space(st["warn_percent"]), cz_settings=st, plan=plan, plan_keep=plan_keep, image_types=config.IMAGE_TYPES, friendly_names=config.FRIENDLY_NAMES, ) @app.route("/backups") def clonezilla_backups(): """Clonezilla image directories and machine-number zips on the share, grouped by machine number (services/clonezilla_images.py).""" return _backups_page() @app.route("/backups/image//delete", methods=["POST"]) def clonezilla_image_delete(name): try: clonezilla_images.delete_image(name) audit("BACKUP_IMAGE_DELETE", name) flash(f"Deleted image {name}.", "success") except (clonezilla_images.ImageError, OSError) as ex: flash(str(ex), "danger") return redirect(url_for("clonezilla_backups")) @app.route("/backups/image//move", methods=["POST"]) def clonezilla_image_move(name): try: new = clonezilla_images.move_image(name, request.form.get("machine", "")) audit("BACKUP_IMAGE_MOVE", f"{name} -> {new}") flash(f"Moved {name} to machine {clonezilla_images.norm_machine(request.form.get('machine'))}: now {new}.", "success") except (clonezilla_images.ImageError, OSError) as ex: flash(str(ex), "danger") return redirect(url_for("clonezilla_backups")) @app.route("/backups/image//tar") def clonezilla_image_tar(name): """The image directory as a tar, streamed file by file (no temp copy).""" try: gen = clonezilla_images.tar_stream(name) except clonezilla_images.ImageError as ex: flash(str(ex), "danger") return redirect(url_for("clonezilla_backups")) audit("BACKUP_IMAGE_DOWNLOAD", name) resp = Response(stream_with_context(gen), mimetype="application/x-tar") resp.headers["Content-Disposition"] = f'attachment; filename="{secure_filename(name) or "image"}.tar"' resp.headers["X-Accel-Buffering"] = "no" return resp @app.route("/backups/settings", methods=["POST"]) def clonezilla_settings(): try: st = clonezilla_images.save_settings(request.form.get("keep_last", ""), request.form.get("warn_percent", "")) audit("BACKUP_SETTINGS", f"keep_last={st['keep_last']} warn_percent={st['warn_percent']}") flash("Clonezilla backup settings saved." + ("" if not st["keep_last"] else " Nothing is deleted until you press Apply retention and confirm."), "success") except clonezilla_images.ImageError as ex: flash(str(ex), "danger") return redirect(url_for("clonezilla_backups")) @app.route("/backups/retention") def clonezilla_retention_preview(): """What 'keep last N per machine' would delete. Shows only; the apply is a separate POST of exactly these names.""" keep = clonezilla_images.settings()["keep_last"] if not keep: flash("Set Keep last N images per machine first.", "warning") return redirect(url_for("clonezilla_backups")) return _backups_page(plan=clonezilla_images.retention_plan(keep), plan_keep=keep) @app.route("/backups/retention/apply", methods=["POST"]) def clonezilla_retention_apply(): keep = clonezilla_images.settings()["keep_last"] names = request.form.getlist("name") if not keep or not names: flash("Nothing to delete.", "warning") return redirect(url_for("clonezilla_backups")) deleted, skipped = clonezilla_images.apply_retention(names, keep) for n in deleted: audit("BACKUP_RETENTION_DELETE", f"{n} (keep last {keep})") msg = f"Retention (keep last {keep}): deleted {len(deleted)} image{'s' if len(deleted) != 1 else ''}." if skipped: msg += f" Kept {len(skipped)} that are no longer due for deletion: {', '.join(skipped[:5])}{'...' if len(skipped) > 5 else ''}." flash(msg, "success" if not skipped else "warning") return redirect(url_for("clonezilla_backups")) @app.route("/backups/upload", methods=["POST"]) def clonezilla_upload(): if "backup_file" not in request.files: flash("No file selected.", "danger") return redirect(url_for("clonezilla_backups")) f = request.files["backup_file"] if not f.filename: flash("No file selected.", "danger") return redirect(url_for("clonezilla_backups")) filename = secure_filename(f.filename) if not filename.lower().endswith(".zip"): flash("Only .zip files are accepted.", "danger") return redirect(url_for("clonezilla_backups")) os.makedirs(config.CLONEZILLA_SHARE, exist_ok=True) dest = os.path.join(config.CLONEZILLA_SHARE, filename) # A backup is a whole disk image: never replace one by accident. # The share is SMB: names differ only in case are the same file to Windows. clash = next((n for n in os.listdir(config.CLONEZILLA_SHARE) if n.lower() == filename.lower()), None) if clash and request.form.get("overwrite") != "1": flash(f"{clash} already exists. Tick Replace the existing backup to overwrite it.", "danger") return redirect(url_for("clonezilla_backups")) if clash: dest = os.path.join(config.CLONEZILLA_SHARE, clash) # replace it, never a case twin f.save(dest) audit("BACKUP_UPLOAD", filename) flash(f"Uploaded {filename} successfully.", "success") return redirect(url_for("clonezilla_backups")) @app.route("/backups/download/") def clonezilla_download(filename): filename = secure_filename(filename) fpath = os.path.join(config.CLONEZILLA_SHARE, filename) if not os.path.isfile(fpath): flash(f"Backup not found: {filename}", "danger") return redirect(url_for("clonezilla_backups")) return send_file(fpath, as_attachment=True) @app.route("/backups/delete/", methods=["POST"]) def clonezilla_delete(filename): filename = secure_filename(filename) fpath = os.path.join(config.CLONEZILLA_SHARE, filename) if os.path.isfile(fpath): os.remove(fpath) audit("BACKUP_DELETE", filename) flash(f"Deleted {filename}.", "success") else: flash(f"Backup not found: {filename}", "danger") return redirect(url_for("clonezilla_backups")) # --------------------------------------------------------------------------- # Routes - Blancco Reports # --------------------------------------------------------------------------- @app.route("/reports") def blancco_reports(): """Every report on the share, newest erasure first, grouped by month. Rows are dated by the timestamp inside the report (the file date resets when a report is copied) and carry per-drive outcomes for the row body.""" from collections import OrderedDict reports = [] state = blancco_evidence.load_state() tz = blancco_evidence._tz() if os.path.isdir(config.BLANCCO_REPORTS): for f in os.listdir(config.BLANCCO_REPORTS): fpath = os.path.join(config.BLANCCO_REPORTS, f) # .part files are reports still being copied in by Blancco. if not os.path.isfile(fpath) or f.startswith(".") or f.endswith(".part"): continue stat = os.stat(fpath) ext = os.path.splitext(f)[1].lower() row = {"filename": f, "size": stat.st_size, "type": ext.lstrip(".").upper() or "FILE", "serial": "", "model": "", "state": "", "cloud": "", "document_id": "", "key": "", "ticked": None, "drives": [], "fallback": False, "parse_error": ""} when = None if ext == ".xml": try: data = blancco_report.parse(fpath) sysinfo = (data.get("hardware") or {}).get("system") or {} if isinstance(sysinfo, list): sysinfo = sysinfo[0] if sysinfo else {} doc = (data.get("meta") or {}).get("document_id", "") or "" key = blancco_evidence.report_key(doc, f) # 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. drives = OrderedDict() for i, er in enumerate(data.get("erasures") or []): tgt = er.get("target") or {} if isinstance(tgt, list): tgt = tgt[0] if tgt else {} d = drives.setdefault(blancco_report._drive_key(er, i), { "model": tgt.get("model", ""), "serial": tgt.get("serial", ""), "capacity": tgt.get("capacity", ""), "interface": tgt.get("interface_type", ""), "attempts": [], "ok": False}) ok = (er.get("state") or "").strip().lower() == "successful" d["ok"] = d["ok"] or ok d["attempts"].append({"ok": ok, "standard": er.get("erasure_standard_name", ""), "elapsed": er.get("elapsed_time", ""), "why": blancco_report.failure_reason(er)}) row.update(serial=sysinfo.get("serial", "") or "", model=sysinfo.get("model", "") or "", state=blancco_report.report_result(data), document_id=doc, key=key, cloud=blancco_evidence.cloud_status(state, doc), ticked=(state["servicenow"].get(key) or {}).get("ts"), drives=list(drives.values()), fallback=any(d["ok"] and any(not a["ok"] for a in d["attempts"]) for d in drives.values())) when = blancco_report.report_datetime(data) except Exception as ex: row["parse_error"] = str(ex)[:200] when = (when or datetime.fromtimestamp(stat.st_mtime, timezone.utc)).astimezone(tz) row["when"] = when reports.append(row) reports.sort(key=lambda r: r["when"], reverse=True) # Same window as the to-do: reports older than 30 days are no longer # asked for in ServiceNow (they show "Not ticked", not "To tick"). cutoff = datetime.now(tz) - timedelta(days=30) for r in reports: r["to_tick"] = bool(r["key"] and not r["ticked"] and r["when"] >= cutoff) months = OrderedDict() for r in reports: months.setdefault(r["when"].strftime("%Y-%m"), []).append(r) evidence_dir = os.path.join(config.BLANCCO_REPORTS, "evidence") uploaded = state.get("evidence_uploaded") or {} groups = [] for m, rows in months.items(): name = blancco_evidence.evidence_name(m) + ".zip" groups.append({"month": m, "label": rows[0]["when"].strftime("%B %Y"), "rows": rows, "failed": sum(1 for r in rows if r["state"] == "Failed"), "pack": name if os.path.isfile(os.path.join(evidence_dir, name)) else "", "uploaded": m in uploaded}) counts = {"failed": sum(1 for r in reports if r["state"] == "Failed"), "cloud": sum(1 for r in reports if r["cloud"] == "failed"), "tick": sum(1 for r in reports if r["to_tick"])} return render_template( "reports.html", groups=groups, total=len(reports), counts=counts, this_month=datetime.now(tz).strftime("%Y-%m"), site=config.BLANCCO_SITE, last_month=blancco_evidence.previous_month(), pack_from=blancco_evidence.month_bounds(blancco_evidence.previous_month())[0].date().isoformat(), pack_to=(blancco_evidence.month_bounds(blancco_evidence.previous_month())[1] - timedelta(days=1)).date().isoformat(), todo=blancco_evidence.todo_counts(), ) REPORT_EXPORT_RANGES = {"1m": 1, "3m": 3, "6m": 6, "12m": 12} def _months_ago(now, months): # Same day-of-month N months back, clamped (Mar 31 - 1m -> Feb 28/29). import calendar y, m = divmod(now.year * 12 + now.month - 1 - months, 12) day = min(now.day, calendar.monthrange(y, m + 1)[1]) return now.replace(year=y, month=m + 1, day=day) @app.route("/reports/export") def blancco_export_reports(): import csv import io import zipfile from datetime import timezone rng = request.args.get("range", "1m") fmt = request.args.get("format", "csv") if rng not in REPORT_EXPORT_RANGES or fmt not in ("csv", "zip"): abort(400) now = datetime.now(timezone.utc) since = _months_ago(now, REPORT_EXPORT_RANGES[rng]) # Select by the date inside each XML report; fall back to mtime only # when a report has no parseable date. PDFs (7.20 writes pdf+xml) ride # along with the XML of the same name. rows, files = [], [] cloud_state = blancco_evidence.load_state() folder = config.BLANCCO_REPORTS names = sorted(os.listdir(folder)) if os.path.isdir(folder) else [] for f in names: fpath = os.path.join(folder, f) if not (f.lower().endswith(".xml") and os.path.isfile(fpath)): continue try: data = blancco_report.parse(fpath) except Exception: data = {} when = blancco_report.report_datetime(data) or datetime.fromtimestamp( os.stat(fpath).st_mtime, timezone.utc) if when < since: continue rows.extend(blancco_report.export_rows( f, data, to_display_tz(when).isoformat(), blancco_evidence.cloud_status(cloud_state, (data.get("meta") or {}).get("document_id", "")))) files.append(f) pdf = os.path.splitext(f)[0] + ".pdf" if pdf in names: files.append(pdf) buf = io.StringIO() writer = csv.DictWriter(buf, fieldnames=blancco_report.EXPORT_COLUMNS) writer.writeheader() writer.writerows(rows) stem = f"blancco-reports-{rng}-{now:%Y%m%d}" audit("REPORT_EXPORT", f"{rng} {fmt} {len(files)} files") if fmt == "csv": return Response( buf.getvalue(), mimetype="text/csv", headers={"Content-Disposition": f"attachment; filename={stem}.csv"}, ) # A year of reports can run to hundreds of MB; spill to disk past 64MB. out = tempfile.SpooledTemporaryFile(max_size=64 * 1024 * 1024) with zipfile.ZipFile(out, "w", zipfile.ZIP_DEFLATED) as zf: zf.writestr(f"{stem}/summary.csv", buf.getvalue()) for f in files: zf.write(os.path.join(folder, f), f"{stem}/{f}") out.seek(0) return send_file(out, mimetype="application/zip", as_attachment=True, download_name=f"{stem}.zip") @app.route("/reports/cloud-status", methods=["POST"]) def blancco_cloud_status(): # CSRF-exempt machine-to-machine endpoint: blancco-progress.py, running # inside Blancco, reports whether each report reached the Blancco cloud. p = request.get_json(silent=True) or {} try: blancco_evidence.record_cloud( p.get("document_id", ""), bool(p.get("sent")), p.get("serial", ""), p.get("detail", "")) except ValueError as ex: return jsonify({"error": str(ex)}), 400 return jsonify({"ok": True}), 200 @app.route("/reports/evidence") def blancco_evidence_pack(): """CMMC evidence pack (ENT-056/158) for a site. Either ?from=&to= (dates, YYYY-MM-DD, inclusive, Eastern time) or ?month=YYYY-MM (the monthly pack).""" from datetime import date as _date first, last = request.args.get("from"), request.args.get("to") if first or last: try: first, last = _date.fromisoformat(first or ""), _date.fromisoformat(last or "") except ValueError: flash("Pick a start and an end date for the evidence pack.", "danger") return redirect(url_for("blancco_reports")) if last < first: first, last = last, first if first.year < 2000 or last.year > 2100: # 9999-12-31 + 1 day overflows the date maths further down. flash("Pick dates between the years 2000 and 2100.", "danger") return redirect(url_for("blancco_reports")) if (last - first).days + 1 > 400: # both ends count flash("An evidence pack covers at most about a year (400 days).", "danger") return redirect(url_for("blancco_reports")) out = tempfile.SpooledTemporaryFile(max_size=64 * 1024 * 1024) count = blancco_evidence.build_evidence_zip(None, out, first=first, last=last) out.seek(0) audit("REPORT_EVIDENCE", f"{first.isoformat()}..{last.isoformat()} {count} reports") return send_file(out, mimetype="application/zip", as_attachment=True, download_name=blancco_evidence.evidence_range_name(first, last) + ".zip") month = request.args.get("month") or blancco_evidence.previous_month() if not re.fullmatch(r"20\d\d-(0[1-9]|1[0-2])", month): abort(400) out = tempfile.SpooledTemporaryFile(max_size=64 * 1024 * 1024) count = blancco_evidence.build_evidence_zip(month, out) out.seek(0) audit("REPORT_EVIDENCE", f"{month} {count} reports") return send_file(out, mimetype="application/zip", as_attachment=True, download_name=blancco_evidence.evidence_name(month) + ".zip") @app.route("/reports/servicenow") def blancco_servicenow(): """Old address of the ServiceNow tick list: it is now the first section of the technician to-do.""" return redirect(url_for("blancco_todo", days=request.args.get("days"), all=request.args.get("all"), _anchor="sn")) @app.route("/reports/servicenow/", methods=["POST"]) def blancco_servicenow_mark(document_id): done = request.form.get("done", "1") == "1" try: blancco_evidence.mark_servicenow(document_id, done) except ValueError: abort(400) audit("REPORT_SERVICENOW", f"{document_id} {'done' if done else 'undone'}") return redirect(url_for("blancco_todo", days=request.args.get("days"), all=request.args.get("all"), _anchor="sn")) def _recently_marked(state, year_items, hours=24): """Attestations marked in the last `hours`, per kind, so the to-do can offer Undo on them: {kind: [{"key", "ts", "label"}]}.""" from datetime import timedelta as _td cutoff = datetime.now(timezone.utc) - _td(hours=hours) by_key = {i["key"]: i for i in year_items} out = {} for kind in blancco_evidence.TODO_KINDS: rows = [] for key, rec in (state.get(kind) or {}).items(): if not isinstance(rec, dict): continue # hand-edited state: nothing to undo try: ts = datetime.fromisoformat(str(rec.get("ts")).replace("Z", "+00:00")) except (ValueError, TypeError): continue if ts.tzinfo is None: ts = ts.replace(tzinfo=timezone.utc) if ts < cutoff: continue it = by_key.get(key) label = (f"{it['serial'] or '-'} {it['model']}".strip() if it else key) rows.append({"key": key, "ts": rec.get("ts"), "label": label}) out[kind] = sorted(rows, key=lambda r: r["ts"], reverse=True) return out @app.route("/reports/todo") def blancco_todo(): """Technician to-do: the playbook steps the server cannot do itself. The ServiceNow tick list is its first section (?days=7|30|90|366, ?all=1 also shows reports already ticked).""" include_done = request.args.get("all") == "1" days = max(1, min(request.args.get("days", 30, type=int), 366)) todo = blancco_evidence.todo() hardware = hardware_coverage.open_gaps() year = blancco_evidence.servicenow_items(days=366, include_done=True) return render_template("reports_todo.html", todo=todo, hardware=hardware, sn_items=blancco_evidence.servicenow_items(days=days, include_done=include_done), days=days, include_done=include_done, recent=_recently_marked(blancco_evidence.load_state(), year), total=todo["total"] + len(hardware), 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'}") anchor = {"destroyed": "fail", "evidence_uploaded": "ev", "cloud_resolved": "cloud"}.get(kind) return redirect(url_for("blancco_todo", _anchor=anchor)) @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("/hardware/seen", methods=["POST"]) def hardware_seen(): # CSRF-exempt machine-to-machine endpoint: WinPE (winpe-status-push.ps1) # reports each machine's model, SKU and BIOS version before imaging. p = request.get_json(silent=True) or {} try: hardware_coverage.record(p) except ValueError as ex: return jsonify({"error": str(ex)}), 400 return jsonify({"ok": True}), 200 @app.route("/hardware") def hardware_page(): """Every model WinPE has seen, with driver-pack and BIOS coverage.""" try: # Files with a newer Dell pack/BIOS in the last Updates > Dell scan. hints = dell_updates.update_hints() except Exception: hints = {"drivers": {}, "bios": {}} return render_template("hardware.html", rows=hardware_coverage.coverage(), images=hardware_coverage.IMAGES, hints=hints) @app.route("/hardware/dismiss", methods=["POST"]) def hardware_dismiss(): model = request.form.get("model", "") done = request.form.get("done", "1") == "1" try: hardware_coverage.dismiss(model, done) except ValueError: abort(400) audit("HARDWARE_DISMISS", f"{model} {'dismissed' if done else 'reopened'}") nxt = request.form.get("next") or "" # Only same-site paths ("/x", not "//host" or "https://..."). if not (nxt.startswith("/") and not nxt.startswith("//") and "\\" not in nxt): nxt = url_for("hardware_page") return redirect(nxt) @app.route("/reports/download/") def blancco_download_report(filename): filename = secure_filename(filename) fpath = os.path.join(config.BLANCCO_REPORTS, filename) if not os.path.isfile(fpath): flash(f"Report not found: {filename}", "danger") return redirect(url_for("blancco_reports")) return send_file(fpath, as_attachment=True) @app.route("/reports/view/") def blancco_view_report(filename): filename = secure_filename(filename) fpath = os.path.join(config.BLANCCO_REPORTS, filename) if not os.path.isfile(fpath): flash(f"Report not found: {filename}", "danger") return redirect(url_for("blancco_reports")) if not filename.lower().endswith(".xml"): flash("Formatted view supports XML reports only.", "warning") return redirect(url_for("blancco_reports")) try: data = blancco_report.parse(fpath) except Exception as ex: flash(f"Failed to parse {filename}: {ex}", "danger") return redirect(url_for("blancco_reports")) # The hero summarises per drive: a drive counts as wiped if any of its # erasures succeeded (Purge failing then Clear succeeding is a pass). outcomes = blancco_report.drive_outcomes(data) state = blancco_evidence.load_state() doc = (data.get("meta") or {}).get("document_id", "") or "" key = blancco_evidence.report_key(doc, filename) failed, note = blancco_evidence.failure_note(data) when = blancco_report.report_datetime(data) return render_template("report_view.html", filename=filename, data=data, result=blancco_report.report_result(data), outcomes=outcomes, cloud=blancco_evidence.cloud_status(state, doc), key=key, ticked=(state["servicenow"].get(key) or {}).get("ts"), destroyed=(state.get("destroyed") or {}).get(key), failed=failed, note=note, when=when.astimezone(blancco_evidence._tz()) if when else None) @app.route("/reports/delete/", methods=["POST"]) def blancco_delete_report(filename): filename = secure_filename(filename) fpath = os.path.join(config.BLANCCO_REPORTS, filename) if os.path.isfile(fpath): os.remove(fpath) audit("REPORT_DELETE", filename) flash(f"Deleted {filename}.", "success") else: flash(f"Report not found: {filename}", "danger") return redirect(url_for("blancco_reports")) # --------------------------------------------------------------------------- # Routes - Imaging Progress # --------------------------------------------------------------------------- @app.route("/imaging") def imaging_dashboard(): show_all = request.args.get("all") == "1" sessions = imaging_status.list_sessions() hidden = 0 if not show_all: sessions, hidden = imaging_status.hide_stale(sessions) # Clear all deletes every session file, hidden finished ones included; # log-only bays have no file. The confirm states that number. clear_count = len(imaging_status.list_sessions(include_inferred=False)) return render_template("imaging.html", sessions=imaging_status.floor_order(sessions), hidden=hidden, clear_count=clear_count, show_all=show_all, window_hours=imaging_status.ACTIVE_WINDOW_HOURS) @app.route("/imaging/tiles") def imaging_tiles_partial(): """HTML fragment of the per-bay tile loop only, used by the dashboard's SSE/polling refresh to swap #imaging-tiles innerHTML without a full page reload.""" sessions = imaging_status.list_sessions() if request.args.get("all") != "1": sessions, _hidden = imaging_status.hide_stale(sessions) sessions = imaging_status.floor_order(sessions) # ?limit=N (the Overview): the first N PCs, log-only bays left out. limit = max(0, request.args.get("limit", 0, type=int) or 0) more = 0 if limit: real = [x for x in sessions if x["floor_state"] != "logs"] more = max(0, len(real) - limit) sessions = real[:limit] return render_template("_imaging_tiles.html", sessions=sessions, more=more, compact=bool(limit)) def _sessions_hash() -> str: """Compact fingerprint of the current session list. Used by the SSE stream to detect changes without sending the full payload. Hashing (serial-or-key, status, stage_index, current_stage, last_updated) covers every field the dashboard renders prominently.""" sessions = imaging_status.list_sessions() h = hashlib.sha256() for s in sessions: key = s.get("serial") or f"{s.get('mac','')}-{s.get('ip','')}" h.update(repr(( key, s.get("source", "client"), s.get("status", ""), s.get("stage_index", 0), s.get("current_stage", ""), s.get("last_updated", ""), s.get("laps_password", "") and "1" or "0", )).encode()) return h.hexdigest()[:16] @app.route("/imaging/stream") def imaging_stream(): """Server-Sent Events stream of session-list change pings. Emits one JSON event every SSE_PING_INTERVAL seconds. When the hash changes from the previously sent value, the client fetches /imaging/tiles and re-renders. A keepalive heartbeat is sent on the same cadence so intermediate proxies don't close the connection. Single-threaded dev server can only serve one SSE client at a time. The live PXE box runs gunicorn with multiple workers (see playbook) so this is fine in production. """ SSE_PING_INTERVAL = 5 # seconds between hash checks SSE_MAX_DURATION = 600 # cap connection length so the worker recycles @stream_with_context def gen(): start = time.time() last = None while time.time() - start < SSE_MAX_DURATION: try: cur = _sessions_hash() except Exception as ex: yield f": error {ex}\n\n" cur = last payload = json.dumps({"hash": cur, "ts": int(time.time())}) yield f"data: {payload}\n\n" last = cur time.sleep(SSE_PING_INTERVAL) headers = { "Cache-Control": "no-cache", "X-Accel-Buffering": "no", # disable nginx/apache proxy buffering "Connection": "keep-alive", } return Response(gen(), mimetype="text/event-stream", headers=headers) @app.route("/imaging/status", methods=["POST"]) def imaging_status_post(): # CSRF-exempt machine-to-machine endpoint; see services/csrf.py exempt list. payload = request.get_json(silent=True) or {} if not isinstance(payload, dict): return jsonify({"error": "body must be a JSON object"}), 400 if not payload.get("serial"): return jsonify({"error": "missing serial"}), 400 if not isinstance(payload.get("serial"), str): return jsonify({"error": "serial must be a string"}), 400 try: state = imaging_status.update_session(payload) except ValueError as ex: return jsonify({"error": str(ex)}), 400 except Exception as ex: audit("IMAGING_STATUS_ERROR", str(ex)) return jsonify({"error": str(ex)}), 500 return jsonify({"ok": True, "serial": state["serial"]}), 200 @app.route("/imaging/.json") def imaging_session_json(serial): serial = secure_filename(serial) s = imaging_status.get_session(serial) if not s: return jsonify({"error": "not found"}), 404 return jsonify(s) @app.route("/imaging/session/") def imaging_session_detail(serial): """Per-bay forensics page: stage timeline, full sidecar log, all session metadata. Linked from the dashboard tile. Returns 404 if no session JSON exists for the serial.""" serial = secure_filename(serial) s = imaging_status.get_session(serial) if not s: flash(f"No session for serial {serial}.", "danger") return redirect(url_for("imaging_dashboard")) full_log, truncated = imaging_status.read_full_log(serial) return render_template( "imaging_detail.html", session=s, full_log=full_log, full_log_truncated=truncated, ) @app.route("/imaging/delete/", methods=["POST"]) def imaging_delete_session(serial): serial = secure_filename(serial) if imaging_status.delete_session(serial): audit("IMAGING_DELETE", serial) flash(f"Cleared imaging session {serial}.", "success") else: flash(f"Session not found: {serial}", "danger") return redirect(url_for("imaging_dashboard")) @app.route("/imaging/delete_all", methods=["POST"]) def imaging_delete_all(): n = imaging_status.delete_all_sessions() audit("IMAGING_DELETE_ALL", str(n)) flash(f"Cleared {n} imaging session(s).", "success") return redirect(url_for("imaging_dashboard")) @app.route("/imaging//laps", methods=["POST"]) def imaging_set_laps(serial): """Save (or clear with empty value) the LAPS password for a bay so it survives the dashboard's 5s auto-refresh. JSON body: {"password": "..."}. Empty string removes the field. Daily reset wipes natural risk.""" serial = secure_filename(serial) body = request.get_json(silent=True) or {} pw = body.get("password", "") if not isinstance(pw, str): return {"ok": False, "error": "password must be string"}, 400 if pw == "": # Clear by direct file write. update_session() merges payload INTO # existing state and skips empty values, so it cannot remove a key. # Pop the laps_password key directly from the session JSON and # write the result atomically. path = imaging_status._path_for(serial) if os.path.isfile(path): try: with open(path, "r") as f: state = json.load(f) except (json.JSONDecodeError, OSError): state = {} if "laps_password" in state: state.pop("laps_password", None) state["last_updated"] = imaging_status._now_iso() fd, tmp = tempfile.mkstemp(dir=config.IMAGING_DIR, prefix=".tmp-", suffix=".json") try: with os.fdopen(fd, "w") as f: json.dump(state, f, indent=2) os.replace(tmp, path) except Exception: try: os.unlink(tmp) except OSError: pass raise return {"ok": True, "cleared": True} # Only for a bay that exists: saving for a cleared session would create a # bare session file with no stage fields. if not os.path.isfile(imaging_status._path_for(serial)): return {"ok": False, "error": "no such session"}, 404 imaging_status.update_session({"serial": serial, "laps_password": pw}) return {"ok": True} # --------------------------------------------------------------------------- # Routes - Enrollment Packages # --------------------------------------------------------------------------- PPKG_CONF = os.path.join(config.ENROLLMENT_SHARE, "ppkg.conf") def read_ppkg_conf(): """Read the active-PPKG config (SOURCE_PPKG/PPKG_VER/PPKG_EXP) that startnet.cmd sources at boot. Returns a dict (empty if unset).""" data = {} if os.path.isfile(PPKG_CONF): try: with open(PPKG_CONF, "r", errors="replace") as fh: for line in fh: if "=" in line: k, _, v = line.strip().partition("=") data[k.strip()] = v.strip() except OSError: pass return data def write_ppkg_conf(source_ppkg, ppkg_ver, ppkg_exp): """Write ppkg.conf (CRLF) so startnet.cmd picks up the active ppkg at boot with no boot.wim edit.""" lines = [ f"SOURCE_PPKG={source_ppkg}", f"PPKG_VER={ppkg_ver}", f"PPKG_EXP={ppkg_exp}", ] with open(PPKG_CONF, "w", newline="\r\n") as fh: fh.write("\n".join(lines) + "\n") @app.route("/enrollment") def enrollment(): packages = [] if os.path.isdir(config.ENROLLMENT_PPKG_DIR): for f in sorted(os.listdir(config.ENROLLMENT_PPKG_DIR)): fpath = os.path.join(config.ENROLLMENT_PPKG_DIR, f) if os.path.isfile(fpath) and f.lower().endswith(".ppkg"): stat = os.stat(fpath) packages.append({ "filename": f, "size": stat.st_size, "modified": stat.st_mtime, }) return render_template( "enrollment.html", packages=packages, mcl_packages=mcl_ppkg.packages(), mcl_archive=mcl_ppkg.archived(), mcl_links=mcl_ppkg.image_links(), active_ppkg=read_ppkg_conf(), effective_ppkg=(read_ppkg_conf().get("SOURCE_PPKG", "").strip().strip('"').strip() or STARTNET_DEFAULT_PPKG), image_types=config.IMAGE_TYPES, friendly_names=config.FRIENDLY_NAMES, ) @app.route("/enrollment/upload", methods=["POST"]) def enrollment_upload(): if "ppkg_file" not in request.files: flash("No file selected.", "danger") return redirect(url_for("enrollment")) f = request.files["ppkg_file"] if not f.filename: flash("No file selected.", "danger") return redirect(url_for("enrollment")) filename = secure_filename(f.filename) if not filename.lower().endswith(".ppkg"): flash("Only .ppkg files are accepted.", "danger") return redirect(url_for("enrollment")) # An MCL name would also match the ppkg.conf regex below and become the # shopfloor package; those go to BPRT through the MCL section instead. # Check the sanitised name too: secure_filename turns spaces into # underscores, which can make an MCL name out of one that was not. if mcl_ppkg.is_mcl_name(f.filename) or mcl_ppkg.is_mcl_name(filename): flash(f"{filename} is a standard/engineer MCL package; upload it in the MCL section.", "danger") return redirect(url_for("enrollment") + "#mcl") os.makedirs(config.ENROLLMENT_PPKG_DIR, exist_ok=True) dest = os.path.join(config.ENROLLMENT_PPKG_DIR, filename) f.save(dest) audit("ENROLLMENT_UPLOAD", filename) flash(f"Uploaded {filename} successfully.", "success") # 2b: set this ppkg active for startnet.cmd by writing ppkg.conf on the # share (no boot.wim edit). Only for the standard GCCH_..._v.ppkg name. m = re.search(r"_(v\d+(?:\.\d+)*)\.ppkg$", filename, re.IGNORECASE) if m: ver = m.group(1) expiry = (request.form.get("expiry") or "").strip() if not expiry: expiry = read_ppkg_conf().get("PPKG_EXP", "") try: write_ppkg_conf(filename, ver, expiry) audit("PPKG_ACTIVATE", f"{filename} ver={ver} exp={expiry}") flash(f"Set {filename} as the active PPKG - startnet.cmd will use it.", "success") except OSError as exc: flash(f"Uploaded, but could not update ppkg.conf: {exc}", "warning") return redirect(url_for("enrollment")) @app.route("/enrollment/mcl/upload", methods=["POST"]) def mcl_ppkg_upload(): """Standard/engineer MCL package: replaces the one with the same tenant/environment/purpose, archiving the old (services.mcl_ppkg).""" f = request.files.get("ppkg_file") if not f or not f.filename: flash("No file selected.", "danger") return redirect(url_for("enrollment") + "#mcl") try: info, moved = mcl_ppkg.install_upload(f, f.filename) except (ValueError, OSError) as exc: flash(f"Upload rejected: {exc}", "danger") return redirect(url_for("enrollment") + "#mcl") name = info["filename"] audit("MCL_PPKG_UPLOAD", f"{name} replaces {', '.join(moved) or 'nothing'}") flash(f"Installed {name} for {info['prefix']}" + (f"; archived {', '.join(moved)}." if moved else "."), "success") for warning in info["warnings"]: flash(warning + ".", "warning") return redirect(url_for("enrollment") + "#mcl") @app.route("/enrollment/mcl/archive", methods=["POST"]) def mcl_ppkg_archive(): """Move one package out of BPRT (clears a duplicate made by hand).""" try: name = mcl_ppkg.archive(request.form.get("filename", "")) except (ValueError, OSError) as exc: flash(f"Archive failed: {exc}", "danger") return redirect(url_for("enrollment") + "#mcl") audit("MCL_PPKG_ARCHIVE", name) flash(f"Archived {name}.", "success") return redirect(url_for("enrollment") + "#mcl") @app.route("/enrollment/mcl/restore", methods=["POST"]) def mcl_ppkg_restore(): batch = request.form.get("batch", "") name = request.form.get("filename", "") try: info, moved = mcl_ppkg.restore(batch, name) except (ValueError, OSError) as exc: flash(f"Restore failed: {exc}", "danger") return redirect(url_for("enrollment") + "#mcl") name = info["filename"] audit("MCL_PPKG_RESTORE", f"{batch}/{name} replaces {', '.join(moved) or 'nothing'}") flash(f"Restored {name}" + (f"; archived {', '.join(moved)}." if moved else "."), "success") return redirect(url_for("enrollment") + "#mcl") @app.route("/enrollment/download/") def enrollment_download(filename): filename = secure_filename(filename) fpath = os.path.join(config.ENROLLMENT_PPKG_DIR, filename) if not os.path.isfile(fpath): flash(f"Package not found: {filename}", "danger") return redirect(url_for("enrollment")) return send_file(fpath, as_attachment=True) # The package startnet.cmd uses when ppkg.conf is missing or names none # (playbook/startnet.cmd: set SOURCE_PPKG=...). STARTNET_DEFAULT_PPKG = "GCCH_Prod_SFLD_v4.16.ppkg" @app.route("/enrollment/delete/", methods=["POST"]) def enrollment_delete(filename): filename = secure_filename(filename) fpath = os.path.join(config.ENROLLMENT_PPKG_DIR, filename) # startnet.cmd falls back to this package when ppkg.conf names none. active = read_ppkg_conf().get("SOURCE_PPKG", "").strip().strip('"').strip() or STARTNET_DEFAULT_PPKG if filename.lower() == active.lower(): # startnet.cmd would keep pointing at a file that no longer exists. flash(f"{filename} is the active package. Upload or activate another one before deleting it.", "danger") return redirect(url_for("enrollment")) if os.path.isfile(fpath): os.remove(fpath) audit("ENROLLMENT_DELETE", filename) flash(f"Deleted {filename}.", "success") else: flash(f"Package not found: {filename}", "danger") return redirect(url_for("enrollment")) # --------------------------------------------------------------------------- # Routes - startnet.cmd Editor (boot.wim) # # The raw full-text editor (POST /startnet/save) is the source of truth and the # always-available fallback. The Settings and Boot-menu tabs are conveniences # that parse the CURRENT extracted content to prefill, then round-trip an # apply_* result back through the SAME wim.update_startnet save path. Every save # path snapshots the prior startnet.cmd first (feature 4) so a bad edit can be # rolled back from the Snapshots tab. # --------------------------------------------------------------------------- def _startnet_wiminfo(): """Parse `wiminfo boot.wim` into a flat dict (empty on any failure).""" import subprocess info = {} try: result = subprocess.run( ["wiminfo", config.BOOT_WIM], capture_output=True, text=True, timeout=15, ) if result.returncode == 0: for line in result.stdout.splitlines(): if ":" in line: key, _, val = line.partition(":") info[key.strip()] = val.strip() except Exception: pass return info def _startnet_current(): """Extracted startnet.cmd text from boot.wim, or '' if unavailable.""" if not os.path.isfile(config.BOOT_WIM): return "" return wim.extract_startnet(config.BOOT_WIM) or "" def _startnet_save_content(new_content, prior, action, detail): """Snapshot `prior` then write `new_content` to boot.wim via wimtools. Central save path shared by the raw editor, the settings tab, the menu tab, and snapshot-restore so EVERY write snapshots first. Flashes success/failure and returns True on success. Lint warnings are surfaced but never block. """ try: wim.save_snapshot(prior, note=action) except Exception as exc: # A snapshot failure should not silently drop the safety net; warn but # still let the operator save (the raw editor is the source of truth). flash(f"Warning: could not snapshot prior startnet.cmd: {exc}", "warning") ok, err = wim.update_startnet(config.BOOT_WIM, new_content) if not ok: flash(f"Failed to update boot.wim: {err}", "danger") return False lint = wim.lint_startnet(new_content) for e in lint.get("errors", []): loc = f"line {e['line']}: " if e.get("line") else "" flash(f"Lint error - {loc}{e['message']}", "warning") for w in lint.get("warnings", []): loc = f"line {w['line']}: " if w.get("line") else "" flash(f"Lint warning - {loc}{w['message']}", "warning") audit(action, detail) return True @app.route("/startnet") def startnet_editor(): wim_exists = os.path.isfile(config.BOOT_WIM) content = "" wim_info = {} settings = {"server_ip": None, "menu_timeout": None, "menu_default": None} boot_menu = [] lint = {"errors": [], "warnings": []} if wim_exists: content = _startnet_current() wim_info = _startnet_wiminfo() settings = wim.parse_settings(content) boot_menu = wim.parse_boot_menu(content) lint = wim.lint_startnet(content) snapshots = wim.list_snapshots() return render_template( "startnet_editor.html", wim_exists=wim_exists, wim_path=config.BOOT_WIM, content=content, wim_info=wim_info, settings=settings, boot_menu=boot_menu, lint=lint, snapshots=snapshots, shopfloor_items=read_shopfloor_menu(), shopfloor_available=available_pctypes(), image_types=config.IMAGE_TYPES, friendly_names=config.FRIENDLY_NAMES, ) @app.route("/startnet/save", methods=["POST"]) def startnet_save(): if not os.path.isfile(config.BOOT_WIM): flash("boot.wim not found.", "danger") return redirect(url_for("startnet_editor")) content = request.form.get("content", "") prior = _startnet_current() if _startnet_save_content(content, prior, "STARTNET_SAVE", "boot.wim updated (raw)"): flash("startnet.cmd updated successfully in boot.wim.", "success") return redirect(url_for("startnet_editor")) @app.route("/startnet/settings", methods=["POST"]) def startnet_save_settings(): """Apply the Settings tab into the CURRENT extracted content, then save. Only server_ip / menu_timeout / menu_default are touched, via targeted replacement (apply_settings never rebuilds the file). Blank fields are left alone. The raw text stays the source of truth: we apply into what is on boot.wim right now, not a rebuilt file. """ if not os.path.isfile(config.BOOT_WIM): flash("boot.wim not found.", "danger") return redirect(url_for("startnet_editor")) prior = _startnet_current() settings = {} ip = (request.form.get("server_ip") or "").strip() if ip: settings["server_ip"] = ip to = (request.form.get("menu_timeout") or "").strip() if to: try: settings["menu_timeout"] = int(to) except ValueError: flash("Menu timeout must be a whole number of seconds.", "danger") return redirect(url_for("startnet_editor")) md = (request.form.get("menu_default") or "").strip() if md: settings["menu_default"] = md new_content = wim.apply_settings(prior, settings) if new_content == prior: flash("No settings changed.", "info") return redirect(url_for("startnet_editor")) if _startnet_save_content(new_content, prior, "STARTNET_SETTINGS", f"settings applied: {settings}"): flash("Startnet settings applied and saved to boot.wim.", "success") return redirect(url_for("startnet_editor")) @app.route("/startnet/menu", methods=["POST"]) def startnet_save_menu(): """Apply the Boot-menu tab into the CURRENT extracted content, then save. The tab posts an `items` JSON array of {label, target, target_image}; list ORDER defines the new 1..N numbering (any 'num' is ignored). apply_boot_menu regenerates only the echo lines, choice count, dispatch, and action blocks; existing block bodies are byte-for-byte preserved. It RAISES ValueError when the structure is ambiguous or a renumber would desync the enrollment %choice% router (e.g. reorder/remove of a cross-referenced image); we catch it, flash the reason, and leave the raw text untouched. """ if not os.path.isfile(config.BOOT_WIM): flash("boot.wim not found.", "danger") return redirect(url_for("startnet_editor")) payload = request.form.get("items", "[]") try: items = json.loads(payload) except json.JSONDecodeError: flash("Invalid boot-menu payload.", "danger") return redirect(url_for("startnet_editor")) if not isinstance(items, list): flash("Boot-menu payload must be a list of items.", "danger") return redirect(url_for("startnet_editor")) for it in items: if not isinstance(it, dict) or not (it.get("target") or "").strip(): flash("Every boot-menu entry needs a non-empty Target.", "danger") return redirect(url_for("startnet_editor")) prior = _startnet_current() try: new_content = wim.apply_boot_menu(prior, items) except ValueError as exc: flash(f"Boot menu not rewritten (kept raw text): {exc}", "danger") return redirect(url_for("startnet_editor")) if new_content == prior: flash("No boot-menu changes to save.", "info") return redirect(url_for("startnet_editor")) if _startnet_save_content(new_content, prior, "STARTNET_MENU", f"boot menu rebuilt ({len(items)} item(s))"): flash("Boot menu rebuilt and saved to boot.wim.", "success") return redirect(url_for("startnet_editor")) @app.route("/startnet/restore", methods=["POST"]) def startnet_restore_snapshot(): """Restore a snapshot back through the save path. Reads the snapshot (path-traversal safe), snapshots the ABOUT-TO-BE-REPLACED current startnet first, then writes the restored bytes to boot.wim. """ if not os.path.isfile(config.BOOT_WIM): flash("boot.wim not found.", "danger") return redirect(url_for("startnet_editor")) snapshot_id = (request.form.get("snapshot_id") or "").strip() restored = wim.read_snapshot(snapshot_id) if restored is None: flash(f"Snapshot not found: {snapshot_id}", "danger") return redirect(url_for("startnet_editor")) prior = _startnet_current() if _startnet_save_content(restored, prior, "STARTNET_RESTORE", f"restored snapshot {snapshot_id}"): flash(f"Restored startnet.cmd from {snapshot_id}.", "success") return redirect(url_for("startnet_editor")) @app.route("/startnet/diff", methods=["POST"]) def startnet_diff(): """Return a unified diff as JSON. Two modes (body may be form or JSON): - snapshot_id -> diff that snapshot (fromfile) vs current boot.wim (tofile) - content -> diff current boot.wim (fromfile) vs pending edit (tofile) JSON callers must send the CSRF token in the X-CSRF-Token header. """ body = request.get_json(silent=True) or {} snapshot_id = (request.form.get("snapshot_id") or body.get("snapshot_id") or "").strip() pending = request.form.get("content") if pending is None: pending = body.get("content") current = _startnet_current() if snapshot_id: snap = wim.read_snapshot(snapshot_id) if snap is None: return jsonify({"error": f"snapshot not found: {snapshot_id}"}), 404 diff = wim.unified_diff(snap, current, fromfile=snapshot_id, tofile="current") return jsonify({"diff": diff, "mode": "snapshot", "snapshot_id": snapshot_id}) if pending is not None: diff = wim.unified_diff(current, pending, fromfile="current", tofile="pending") return jsonify({"diff": diff, "mode": "pending"}) return jsonify({"error": "provide snapshot_id or content"}), 400 # --------------------------------------------------------------------------- # Routes - Shopfloor boot-menu (data-driven menu.json) # --------------------------------------------------------------------------- SHOPFLOOR_SETUP_DIR = os.path.join(config.ENROLLMENT_SHARE, "shopfloor-setup") SHOPFLOOR_MENU_JSON = os.path.join(SHOPFLOOR_SETUP_DIR, "menu.json") def read_shopfloor_menu(): """The shopfloor PC-type menu items from menu.json (list; [] if absent).""" try: with open(SHOPFLOOR_MENU_JSON) as fh: return json.load(fh).get("shopfloor", []) except (OSError, json.JSONDecodeError): return [] def available_pctypes(): """gea-shopfloor-* handler dirs on the share - the only valid menu keys.""" try: return sorted( d for d in os.listdir(SHOPFLOOR_SETUP_DIR) if d.startswith("gea-shopfloor-") and os.path.isdir(os.path.join(SHOPFLOOR_SETUP_DIR, d)) ) except OSError: return [] @app.route("/shopfloor-menu", methods=["GET", "POST"]) def shopfloor_menu(): if request.method == "POST": try: items = json.loads(request.form.get("items", "[]")) except json.JSONDecodeError: flash("Invalid menu payload.", "danger") return redirect(url_for("startnet_editor") + "#tab-shopfloor") valid = set(available_pctypes()) clean = [] for it in items: if not isinstance(it, dict): continue key = (it.get("key") or "").strip() label = (it.get("label") or "").strip() if not key or not label: continue if key not in valid: flash(f"Skipped '{label}': {key} has no shopfloor-setup handler.", "warning") continue clean.append({ "key": key, "label": label, "hint": (it.get("hint") or "").strip(), "enabled": bool(it.get("enabled", True)), }) if not clean: flash("Refused to save an empty menu.", "danger") return redirect(url_for("startnet_editor") + "#tab-shopfloor") try: os.makedirs(SHOPFLOOR_SETUP_DIR, exist_ok=True) with open(SHOPFLOOR_MENU_JSON, "w") as fh: json.dump({"shopfloor": clean}, fh, indent=2) audit("SHOPFLOOR_MENU_SAVE", f"{len(clean)} items") flash("Shopfloor boot menu saved - WinPE reads it at next boot.", "success") except OSError as exc: flash(f"Could not write menu.json: {exc}", "danger") return redirect(url_for("startnet_editor") + "#tab-shopfloor") return redirect(url_for("startnet_editor") + "#tab-shopfloor") # --------------------------------------------------------------------------- # Routes - Audit Log # --------------------------------------------------------------------------- @app.route("/audit") def audit_log(): entries = [] if os.path.isfile(config.AUDIT_LOG): with open(config.AUDIT_LOG, "r") as fh: for line in fh: entries.append(line.strip()) entries.reverse() return render_template( "audit.html", entries=entries, image_types=config.IMAGE_TYPES, friendly_names=config.FRIENDLY_NAMES, ) # --------------------------------------------------------------------------- # Routes - JSON API # --------------------------------------------------------------------------- @app.route("/api/services") def api_services(): services = {s: system.service_status(s) for s in ("dnsmasq", "apache2", "smbd")} return jsonify(services) @app.route("/api/images") def api_images(): image_list = [images.image_status(it) for it in config.IMAGE_TYPES] return jsonify(image_list) @app.route("/api/images//unattend", methods=["POST"]) def api_save_unattend(image_type): if image_type not in config.IMAGE_TYPES: return jsonify({"error": "Unknown image type"}), 404 xml_file = fs.unattend_path(image_type) payload = request.get_json(silent=True) if not payload or not isinstance(payload, dict): return jsonify({"error": "No JSON body provided"}), 400 # {"raw_xml": ...} writes the text as is; any other body is form data # (keys left out stay as they are in the file). Backup, undo point and # lint gate as on the Unattend tab. if "raw_xml" in payload: body = {"mode": "raw", "raw_xml": payload["raw_xml"]} else: body = {"mode": "form", "form": payload} try: plan = images_unattend_views.save(image_type, body, action="Unattend save (API)") except images_unattend_views.Refused as exc: return jsonify({"error": str(exc)}), exc.status if plan["changed"]: audit("UNATTEND_SAVE_API", image_type) return jsonify({"status": "ok", "path": xml_file, "changed": plan["changed"]}) # --------------------------------------------------------------------------- # Template helpers # --------------------------------------------------------------------------- def to_display_tz(value): """datetime (naive = UTC), ISO-8601 string or Unix timestamp -> aware datetime in config.DISPLAY_TZ; None when it cannot be read.""" from datetime import timezone from zoneinfo import ZoneInfo if value is None or value == "": return None if isinstance(value, (int, float)): dt = datetime.fromtimestamp(value, timezone.utc) elif isinstance(value, datetime): dt = value if value.tzinfo else value.replace(tzinfo=timezone.utc) else: try: dt = datetime.fromisoformat(str(value).replace("Z", "+00:00")) except ValueError: return None if not dt.tzinfo: dt = dt.replace(tzinfo=timezone.utc) return dt.astimezone(ZoneInfo(config.DISPLAY_TZ)) @app.template_filter("local_dt") def local_dt(value, seconds=False): """Any timestamp shown in the UI, 12-hour, as e.g. '2026-10-05 1:54 PM EDT'.""" dt = to_display_tz(value) if dt is None: return value or "-" return dt.strftime("%Y-%m-%d %-I:%M:%S %p %Z" if seconds else "%Y-%m-%d %-I:%M %p %Z") @app.template_filter("timestamp_fmt") def timestamp_fmt(ts): """Format a Unix timestamp in the display timezone.""" return local_dt(ts) @app.template_filter("tojson_ordered") def tojson_ordered(value): """JSON in the data's own key order (Flask's tojson sorts keys, which turns a save of an unchanged row into a reordered-keys diff). Output is a plain string, so it is HTML-escaped like any other value.""" return json.dumps(value, ensure_ascii=False) @app.template_filter("short_date") def short_date(value): """ISO timestamp -> '14 Oct' in the site's time zone; '' if unreadable.""" try: dt = datetime.fromisoformat(str(value).replace("Z", "+00:00")) except ValueError: return "" if dt.tzinfo is None: dt = dt.replace(tzinfo=timezone.utc) return dt.astimezone(blancco_evidence._tz()).strftime("%d %b") @app.template_filter("month_label") def month_label(value): """'2026-09' -> 'September 2026'; anything else unchanged.""" try: return datetime.strptime(str(value), "%Y-%m").strftime("%B %Y") except ValueError: return value # Report page: why an erasure failed, straight from the report. app.jinja_env.globals["blancco_failure_reason"] = blancco_report.failure_reason @app.context_processor def inject_globals(): return { "all_image_types": config.IMAGE_TYPES, "all_friendly_names": config.FRIENDLY_NAMES, # image pages: tab table, header data, tab URLs (see _image_header.html) "image_tabs": image_pages.IMAGE_TABS, "image_header_info": image_pages.header_info, "image_tab_url": image_pages.tab_url, "image_tab_for_endpoint": image_pages.tab_for_endpoint, "blancco_todo": dict(blancco_evidence.todo_counts(), hardware=hardware_coverage.gap_count(), total=blancco_evidence.todo_counts()["total"] + hardware_coverage.gap_count()), } # --------------------------------------------------------------------------- # Main # --------------------------------------------------------------------------- if __name__ == "__main__": app.run(host="127.0.0.1", port=9010, debug=False, threaded=True)