"""Updates > Blancco: install a new Blancco Drive Eraser release, test first. Upload (or pick) the vendor ISO / USB image, stage it as a background job into WEB_ROOT/blancco/staged//, boot it from a test entry in the iPXE menu, then activate it (backup + atomic swap) or roll back. Staging is a background job (systemd-run) whose progress JSON the page polls; the rest are renames and small file writes and run inline. Logic lives in services/blancco_update.py; this module is the route surface. """ from flask import (Blueprint, abort, flash, jsonify, redirect, render_template, request, send_file, url_for) from services import blancco_update as bu from services.audit import audit bp = Blueprint("blancco_update", __name__, url_prefix="/updates/blancco") def _back(): return redirect(url_for("blancco_update.page")) @bp.route("/") def page(): job = bu.job_state() return render_template( "updates_blancco.html", current=bu.current(), staged=bu.staged(), backups=bu.backups(), sources=bu.sources(), job=job, job_running=job.get("state") == "running", test=bu.test_entry(), initrd=bu.initrd_supports_base(), upload_dir=bu.upload_dir(), source_dirs=bu.source_dirs(), server=bu.SERVER, ) @bp.route("/status.json") def status(): job = bu.job_state() return jsonify({k: job.get(k) for k in ("state", "step", "percent", "message", "log", "version", "source", "started", "finished")}) @bp.route("/upload", methods=["POST"]) def upload(): """Raw body (the page's fetch upload, name in ?name=) streams straight to disk; a plain multipart form works too.""" f = request.files.get("release_file") try: if f is not None: name, size = bu.save_upload(f.stream, f.filename) else: name, size = bu.save_upload(request.stream, request.args.get("name", ""), request.content_length) except (ValueError, RuntimeError, OSError) as ex: if f is None: return jsonify({"ok": False, "error": str(ex)}), 400 flash(f"Upload failed: {ex}", "danger") return _back() audit("BLANCCO_UPLOAD", f"{name} ({size} bytes)") if f is None: return jsonify({"ok": True, "name": name, "size": size}) flash(f"Uploaded {name}.", "success") return _back() @bp.route("/stage", methods=["POST"]) def stage(): key, _, name = request.form.get("source", "").partition(":") try: path = bu.resolve_source(key, name) bu.start_stage(path) except (ValueError, RuntimeError, OSError) as ex: flash(f"Staging not started: {ex}", "danger") return _back() audit("BLANCCO_STAGE", path) flash(f"Staging {name}. This takes a minute or two; the page follows the job.", "info") return _back() @bp.route("/cancel", methods=["POST"]) def cancel(): bu.cancel_stage() audit("BLANCCO_STAGE_CANCEL", "") flash("Staging cancelled; partial files removed.", "warning") return _back() @bp.route("/prefs/", methods=["POST"]) def merged_prefs(version): f = request.files.get("prefs_file") if not f or not f.filename: flash("Choose the merged preferences.xml first.", "danger") return _back() try: meta = bu.save_merged_prefs(version, f.read(4 << 20)) except bu.StageBlocked as ex: flash(str(ex), "danger") return _back() except (ValueError, RuntimeError, OSError) as ex: flash(f"Preferences not accepted: {ex}", "danger") return _back() audit("BLANCCO_PREFS_MERGE", f"{version} preferences {meta.get('prefs_version')}") flash(f"Merged preferences accepted; Blancco {version} is staged and ready to test. " "Commit the same file to playbook/blancco-preferences.xml so a playbook run " "does not put the old one back.", "success") return _back() @bp.route("/vendor-prefs/.xml") def vendor_prefs(version): try: path = bu.vendor_prefs_file(version) except ValueError: abort(404) if not path: abort(404) return send_file(path, mimetype="application/xml", as_attachment=True, download_name=f"vendor-preferences-{version}.xml") @bp.route("/test/", methods=["POST"]) def test_add(version): try: backup = bu.add_test_entry(version) except (ValueError, RuntimeError, OSError) as ex: flash(f"Test entry not added: {ex}", "danger") return _back() audit("BLANCCO_TEST_ENTRY_ADD", f"{version} (menu backup {backup})") flash(f"Boot menu now has \"Blancco Drive Eraser {version} (STAGED - TEST)\". " "PXE boot a test machine and pick it.", "success") return _back() @bp.route("/test/remove", methods=["POST"]) def test_remove(): try: backup = bu.remove_test_entry() except (RuntimeError, OSError) as ex: flash(f"Test entry not removed: {ex}", "danger") return _back() audit("BLANCCO_TEST_ENTRY_REMOVE", f"menu backup {backup}" if backup else "none present") flash("Test entry removed from the boot menu." if backup else "No test entry in the menu.", "success") return _back() @bp.route("/activate/", methods=["POST"]) def activate(version): try: backup, files, removed = bu.activate(version) except (ValueError, RuntimeError, OSError) as ex: flash(f"Not activated: {ex}", "danger") return _back() audit("BLANCCO_ACTIVATE", f"{version}; previous release saved in {backup}") flash(f"Blancco {version} is live ({len(files)} files swapped). The previous release is in " f"{backup}." + (" The test menu entry was removed." if removed is True else ""), "success") if isinstance(removed, str): flash(f"Blancco {version} is live, but {removed}.", "warning") return _back() @bp.route("/rollback/", methods=["POST"]) def rollback(name): try: backup, files = bu.rollback(name) except (ValueError, RuntimeError, OSError) as ex: flash(f"Not rolled back: {ex}", "danger") return _back() audit("BLANCCO_ROLLBACK", f"restored {name}; the replaced release saved in {backup}") flash(f"Restored {name} ({len(files)} files); its folder is released now that the files are " f"live. What was live before is saved in {backup}.", "success") return _back() @bp.route("/staged//delete", methods=["POST"]) def delete_staged(version): try: bu.delete_staged(version) except (ValueError, RuntimeError, OSError) as ex: flash(f"Not deleted: {ex}", "danger") return _back() audit("BLANCCO_DELETE_STAGED", version) flash(f"Deleted staged {version}.", "success") return _back() @bp.route("/backup//delete", methods=["POST"]) def delete_backup(name): try: bu.delete_backup(name) except (ValueError, OSError) as ex: flash(f"Not deleted: {ex}", "danger") return _back() audit("BLANCCO_DELETE_BACKUP", name) flash(f"Deleted {name}.", "success") return _back() @bp.route("/uploads//delete", methods=["POST"]) def delete_upload(name): try: bu.delete_upload(name) except (ValueError, OSError) as ex: flash(f"Not deleted: {ex}", "danger") return _back() audit("BLANCCO_DELETE_UPLOAD", name) flash(f"Deleted {name}.", "success") return _back()