dfd57bef17
- Report view: the hero summarised only the first erasure, so a drive whose Purge failed and Clear fallback succeeded showed Failed while the list said Successful. It now uses the per-drive outcome (report_result) and lists the standards that wiped each drive. - To-do and ServiceNow: reports with no usable document id could never be ticked or cleared (400). They are now tracked by file name (file:<name>). - Imaging: the LAPS password field is masked, with a Show/Hide toggle. - Imaging: succeeded sessions not updated for ACTIVE_WINDOW_HOURS are hidden from the dashboard (failed and in-progress always stay), with Show all. - Enrollment: the active SFLD package can no longer be deleted, which left ppkg.conf pointing at a missing file; its row is marked Active. - Import: confirm before importing, saying upload-folder sources are moved and same-name files in the image are overwritten.
1674 lines
65 KiB
Python
1674 lines
65 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
|
|
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]
|
|
services = [system.service_status(s) for s in ("dnsmasq", "apache2", "smbd")]
|
|
return render_template(
|
|
"dashboard.html",
|
|
images=image_list,
|
|
services=services,
|
|
image_types=config.IMAGE_TYPES,
|
|
friendly_names=config.FRIENDLY_NAMES,
|
|
)
|
|
|
|
|
|
@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():
|
|
reports = []
|
|
state = blancco_evidence.load_state()
|
|
if os.path.isdir(config.BLANCCO_REPORTS):
|
|
for f in sorted(os.listdir(config.BLANCCO_REPORTS), reverse=True):
|
|
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()
|
|
# Surface BIOS serial + system model so operators can find a
|
|
# report for a specific bay without opening each one. Parse
|
|
# cost is one XML walk per .xml report (~50KB-3MB each;
|
|
# negligible at fleet sizes).
|
|
serial = ""
|
|
model = ""
|
|
result = ""
|
|
cloud = ""
|
|
if ext == ".xml":
|
|
try:
|
|
data = blancco_report.parse(fpath)
|
|
sysinfo = (data.get("hardware") or {}).get("system") or {}
|
|
serial = sysinfo.get("serial", "") or ""
|
|
model = sysinfo.get("model", "") or ""
|
|
# Overall result per drive: Failed if any drive was not
|
|
# wiped. A failed Purge followed by a successful Clear
|
|
# fallback on the same drive is a successful wipe.
|
|
result = blancco_report.report_result(data)
|
|
cloud = blancco_evidence.cloud_status(
|
|
state, (data.get("meta") or {}).get("document_id", ""))
|
|
except Exception:
|
|
pass
|
|
reports.append({
|
|
"filename": f,
|
|
"size": stat.st_size,
|
|
"modified": stat.st_mtime,
|
|
"type": ext.lstrip(".").upper() or "FILE",
|
|
"serial": serial,
|
|
"model": model,
|
|
"state": result,
|
|
"cloud": cloud,
|
|
})
|
|
return render_template(
|
|
"reports.html",
|
|
reports=reports,
|
|
site=config.BLANCCO_SITE,
|
|
last_month=blancco_evidence.previous_month(),
|
|
image_types=config.IMAGE_TYPES,
|
|
friendly_names=config.FRIENDLY_NAMES,
|
|
)
|
|
|
|
|
|
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): 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():
|
|
"""What to tick on each asset's Onsite Bin Management RITM (ENT-045/159)."""
|
|
include_done = request.args.get("all") == "1"
|
|
days = request.args.get("days", 30, type=int)
|
|
items = blancco_evidence.servicenow_items(days=max(1, min(days, 366)), include_done=include_done)
|
|
return render_template("reports_servicenow.html", items=items,
|
|
include_done=include_done, days=days)
|
|
|
|
|
|
@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_servicenow", all=request.args.get("all")))
|
|
|
|
|
|
@app.route("/reports/todo")
|
|
def blancco_todo():
|
|
"""Technician to-do: the playbook steps the server cannot do itself."""
|
|
return render_template("reports_todo.html", todo=blancco_evidence.todo(),
|
|
hardware=hardware_coverage.open_gaps(),
|
|
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'}")
|
|
return redirect(url_for("blancco_todo"))
|
|
|
|
|
|
@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)
|
|
return render_template("report_view.html", filename=filename, data=data,
|
|
result=blancco_report.report_result(data), outcomes=outcomes)
|
|
|
|
|
|
@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=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)
|
|
return render_template("_imaging_tiles.html", sessions=sessions)
|
|
|
|
|
|
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)
|
|
|
|
|
|
@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)
|