a6f95fcca1
Reports (/reports):
- Grouped by month of the erasure date inside each report (not the file
date), newest first; older months fold. Each month header shows the
evidence pack state (builds on the 1st, to upload with Download, or
uploaded).
- One line per report: erasure date, system serial, model, drives,
result ('Successful, fallback' when a failed attempt fell back to
another standard), cloud and ServiceNow state. The row body lists every
drive and erasure attempt with the failure reason.
- Search by system serial, model, drive serial or file; filters for
failed, not in cloud and to tick. 'To tick' uses the to-do's 30-day
window; older unticked reports read 'Not ticked'.
- Export and the evidence pack download moved into header popovers.
Report (/reports/view/<file>):
- Header with serial, model and the per-drive result; a notice names
the drive to destroy until its COD is recorded; Copy ServiceNow note.
- One card per drive with every attempt, a step track and the failure
reason; steps and error counts fold. System, customer and provenance
(integrity signature folded, copyable) on the side. Sizes in decimal
units, as on the list.
- blancco_evidence.failure_note() is shared with the ServiceNow list.
To-do (/reports/todo):
- The ServiceNow tick list is now its first section (period and show
ticked filters); /reports/servicenow redirects there.
- Five counters; one line per item with details in the row body.
- COD recorded, Mark uploaded and Resolved confirm first (the attribute
quoting is safe for JSON) and can be undone for 24 hours from the
page. The header total includes hardware gaps, matching the sidebar.
1801 lines
73 KiB
Python
1801 lines
73 KiB
Python
#!/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,
|
|
hardware_coverage,
|
|
blancco_report,
|
|
dell_updates,
|
|
deploy,
|
|
driver_map,
|
|
fs,
|
|
control_files,
|
|
image_pages,
|
|
image_registry,
|
|
images,
|
|
imaging_status,
|
|
mcl_ppkg,
|
|
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
|
|
# ---------------------------------------------------------------------------
|
|
|
|
@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")]
|
|
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,
|
|
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
|
|
scan = dell_updates.last_scan()
|
|
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}
|
|
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}
|
|
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/<image_type>/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/<image_type>/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/<image_type>/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-<ts>-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/<image_type>/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/<image_type>/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/<image_type>/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/<image_type>/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/<image_type>/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/<image_type>/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/<image_type>/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
|
|
# ---------------------------------------------------------------------------
|
|
|
|
@app.route("/backups")
|
|
def clonezilla_backups():
|
|
backups = []
|
|
if os.path.isdir(config.CLONEZILLA_SHARE):
|
|
for f in sorted(os.listdir(config.CLONEZILLA_SHARE)):
|
|
fpath = os.path.join(config.CLONEZILLA_SHARE, f)
|
|
if os.path.isfile(fpath) and f.lower().endswith(".zip"):
|
|
stat = os.stat(fpath)
|
|
backups.append({
|
|
"filename": f,
|
|
"machine": os.path.splitext(f)[0],
|
|
"size": stat.st_size,
|
|
"modified": stat.st_mtime,
|
|
})
|
|
return render_template(
|
|
"backups.html",
|
|
backups=backups,
|
|
image_types=config.IMAGE_TYPES,
|
|
friendly_names=config.FRIENDLY_NAMES,
|
|
)
|
|
|
|
|
|
@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)
|
|
f.save(dest)
|
|
audit("BACKUP_UPLOAD", filename)
|
|
flash(f"Uploaded {filename} successfully.", "success")
|
|
return redirect(url_for("clonezilla_backups"))
|
|
|
|
|
|
@app.route("/backups/download/<filename>")
|
|
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/<filename>", 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(),
|
|
todo=blancco_evidence.todo_counts(),
|
|
)
|
|
|
|
|
|
@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): one ZIP per site and month."""
|
|
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/<document_id>", 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():
|
|
try:
|
|
ts = datetime.fromisoformat(str((rec or {}).get("ts")))
|
|
except ValueError:
|
|
continue
|
|
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/<kind>/<key>", 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/<filename>")
|
|
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'}")
|
|
return redirect(request.form.get("next") or url_for("hardware_page"))
|
|
|
|
|
|
@app.route("/reports/download/<filename>")
|
|
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/<filename>")
|
|
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/<filename>", 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)
|
|
return render_template("imaging.html", sessions=imaging_status.floor_order(sessions), hidden=hidden,
|
|
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 = request.args.get("limit", type=int)
|
|
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 payload.get("serial"):
|
|
return jsonify({"error": "missing serial"}), 400
|
|
try:
|
|
state = imaging_status.update_session(payload)
|
|
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/<serial>.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/<serial>")
|
|
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/<serial>", 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/<serial>/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}
|
|
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(),
|
|
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<ver>.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/<filename>")
|
|
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)
|
|
|
|
|
|
@app.route("/enrollment/delete/<filename>", methods=["POST"])
|
|
def enrollment_delete(filename):
|
|
filename = secure_filename(filename)
|
|
fpath = os.path.join(config.ENROLLMENT_PPKG_DIR, filename)
|
|
active = read_ppkg_conf().get("SOURCE_PPKG", "")
|
|
if active and 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/<image_type>/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)
|
|
|
|
|
|
# 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)
|