pxe-server/webapp/views/images_unattend.py

"""GE Aerospace image pages: the Unattend tab's review, save and undo.

The page itself is unattend_editor in app.py (the "unattend" row of
services/image_pages.IMAGE_TABS, URL /images/<t>/unattend); it renders
templates/unattend_editor.html with page_context() from here. Saving is a
two-step JSON flow driven by static/unattend-editor.js:

  POST /images/<t>/unattend/preview  {mode, form | raw_xml, base_sha}
       -> {files: [control_files.public(plan)], lint, xml}; nothing written
  POST /images/<t>/unattend/save     same body + expect {name: sha}
       -> {ok: true} and a flash; the page reloads
  POST /images/<t>/unattend/undo     (form post) puts the last save back

mode "form" writes the form's values INTO the current file text
(services/unattend.apply_form: comments, *arch*, attribute layout and every
untouched line stay as they are); mode "raw" writes the Raw XML tab as is.
Writes go through services/control_files (FlatUnattendW10.xml.bak-<ts>-webui
next to the file, atomic, owner and mode kept, undo of the last save, scope
"unattend"). A save that adds a lint ERROR (scripts/lint-unattend.py
checks: one over-long command invalidates the whole answer file for its
pass) is refused.

save() is also used by the old form post in unattend_editor and by
/api/images/<t>/unattend, so every way of writing this file gets the same
backup, undo point and lint gate.
"""

import glob
import os
import threading

from flask import Blueprint, flash, jsonify, redirect, request, url_for
from lxml import etree

import config
from services import control_files, fs, image_pages, unattend
from services.audit import audit

bp = Blueprint("images_unattend", __name__, url_prefix="/images")

UNDO_SCOPE = "unattend"
FILE_NAME = "FlatUnattendW10.xml"

# One write at a time (the app serves requests on threads): the stale check,
# backup, write and undo record of one save must not interleave with another
# save or an undo, or the second would silently overwrite the first.
_write_lock = threading.Lock()


class Refused(Exception):
    """A save or preview that cannot go ahead; carries the HTTP status."""

    def __init__(self, message, status=400):
        super().__init__(message)
        self.status = status


def page_context(image_type):
    """Template values for the Unattend tab besides the parsed data."""
    path = fs.unattend_path(image_type)
    exists = os.path.isfile(path)
    text = control_files.read_text(path) if exists else ""
    st = os.stat(path) if exists else None
    preferred = image_pages.preferred_unattend_path(image_type)
    return {
        "file_exists": exists,
        "file_sha": control_files.file_sha256(path),
        "file_mtime": st.st_mtime if st else None,
        "file_size": st.st_size if st else None,
        "backup_count": len(glob.glob(glob.escape(path) + ".bak-*")) if exists else 0,
        "lint": unattend.lint_text(text) if exists else [],
        "last_save": control_files.last_save(image_type, UNDO_SCOPE),
        "unattend_override": (os.path.relpath(preferred, fs.image_root(image_type))
                              if os.path.isfile(preferred) else None),
        "limits": {"path": unattend.MAX_PATH, "commandline": unattend.MAX_COMMANDLINE,
                   "description": unattend.MAX_DESCRIPTION},
    }


def _new_text(current, body):
    """The text body asks to write, from the form or the Raw XML tab."""
    mode = body.get("mode")
    if mode == "raw":
        raw = body.get("raw_xml")
        if not isinstance(raw, str):
            raise Refused("raw_xml must be text")
        try:
            etree.fromstring(raw.lstrip("\ufeff").encode("utf-8"))
        except etree.XMLSyntaxError as exc:
            raise Refused(f"Invalid XML: {exc}")
        # Keep the file's own line endings and BOM: a textarea hands back
        # LF (a plain form post sends CRLF), and a browser may drop a
        # leading BOM. The diff (and the file) then only show what was edited.
        bom = raw.startswith("\ufeff")
        raw = raw.lstrip("\ufeff").replace("\r\n", "\n")
        if "\r\n" in current:
            raw = raw.replace("\n", "\r\n")
        if bom or current.startswith("\ufeff"):
            raw = "\ufeff" + raw
        return raw
    if mode == "form":
        form = body.get("form")
        if not isinstance(form, dict):
            raise Refused("form must be an object")
        if "\ufffd" in current:
            # read_text replaced bytes that are not UTF-8; writing the form
            # into that text would change those bytes on lines nobody edited.
            raise Refused(f"{FILE_NAME} holds bytes that are not UTF-8, so the form cannot "
                          "edit it without changing them. Fix the file in Raw XML first.")
        try:
            return unattend.apply_form(current, form)
        except unattend.ApplyError as exc:
            raise Refused(str(exc))
    raise Refused("mode must be form or raw")


def plan(image_type, body):
    """(control_files plan, lint comparison) for a request body."""
    if not isinstance(body, dict):
        raise Refused("expected a JSON object")
    path = fs.unattend_path(image_type)
    base = body.get("base_sha")
    if base and base != control_files.file_sha256(path):
        raise Refused(f"{FILE_NAME} changed on disk since this page was opened. "
                      "Reload the page (your edits here will be lost) and make them again.", 409)
    current = control_files.read_text(path)
    new = _new_text(current, body)
    return control_files.plan_text(path, new, FILE_NAME), unattend.lint_compare(current, new)


def save(image_type, body, action):
    """Plan, lint-gate, back up, write and record the undo point. Returns
    the plan (plan["changed"] False when there was nothing to write)."""
    with _write_lock:
        p, lint = plan(image_type, body)
        if p["changed"] and lint["new_errors"]:
            msgs = "; ".join(f["message"] for f in lint["findings"] if f["level"] == "ERROR" and f["new"])
            raise Refused(f"Not saved: this change would break the answer file. {msgs}")
        expect = body.get("expect") if isinstance(body.get("expect"), dict) else None
        if not expect:
            # Callers without a preview (the API, the old form post): the
            # file must still be what this plan was built from when it is
            # written (commit checks it under its own lock).
            expect = {p["name"]: p["sha_before"]}
        try:
            control_files.commit(image_type, UNDO_SCOPE, [p], action=action, expect=expect)
        except control_files.Conflict as exc:
            raise Refused(str(exc), 409)
        except OSError as exc:
            raise Refused(f"write failed: {exc}", 500)
        return p


def save_and_flash(image_type, body, action):
    """save() for the form-post routes: audit and flash the outcome."""
    try:
        p = save(image_type, body, action)
    except Refused as exc:
        flash(str(exc), "danger")
        return None
    if p["changed"]:
        audit("UNATTEND_SAVE", f"{image_type} ({body.get('mode')})")
        flash(f"Saved {FILE_NAME}. The previous version is next to it "
              f"({FILE_NAME}.bak-...-webui) and Undo puts it back.", "success")
    else:
        flash(f"{FILE_NAME}: nothing changed, nothing written.", "info")
    return p


def _unknown_json():
    return jsonify({"error": "Unknown image type"}), 404


@bp.route("/<image_type>/unattend/preview", methods=["POST"])
def preview(image_type):
    if image_type not in config.IMAGE_TYPES:
        return _unknown_json()
    try:
        p, lint = plan(image_type, request.get_json(silent=True))
    except Refused as exc:
        return jsonify({"error": str(exc)}), exc.status
    return jsonify({"files": [control_files.public(p)], "lint": lint, "xml": p["new"],
                    "changed": p["changed"]})


@bp.route("/<image_type>/unattend/save", methods=["POST"])
def save_route(image_type):
    if image_type not in config.IMAGE_TYPES:
        return _unknown_json()
    body = request.get_json(silent=True)
    try:
        p = save(image_type, body, action=f"Unattend save ({(body or {}).get('mode')})")
    except Refused as exc:
        return jsonify({"error": str(exc)}), exc.status
    if p["changed"]:
        audit("UNATTEND_SAVE", f"{image_type} ({body.get('mode')})")
        flash(f"Saved {FILE_NAME}. The previous version is next to it "
              f"({FILE_NAME}.bak-...-webui) and Undo puts it back.", "success")
    else:
        flash(f"{FILE_NAME}: nothing changed, nothing written.", "info")
    return jsonify({"ok": True, "changed": p["changed"]})


@bp.route("/<image_type>/unattend/undo", methods=["POST"])
def undo(image_type):
    if image_type not in config.IMAGE_TYPES:
        flash("Unknown image type.", "danger")
        return redirect(url_for("dashboard"))
    try:
        with _write_lock:
            rec = control_files.undo(image_type, UNDO_SCOPE)
        audit("UNATTEND_UNDO", f"{image_type} (save of {rec.get('ts')})")
        flash(f"Undid the last save: {FILE_NAME} is back to how it was before it. "
              f"The undone version is kept as {FILE_NAME}.bak-...-webui-undo.", "success")
    except control_files.Conflict as exc:
        flash(f"Undo refused: {exc}", "danger")
    except OSError as exc:
        flash(f"Undo failed: {exc}", "danger")
    return redirect(image_pages.tab_url(image_type, "unattend"))