pxe-server/scripts/lint-unattend.py

#!/usr/bin/env python3
"""
lint-unattend.py - Validate unattend answer files before they reach a bay.

Well-formed XML is not enough. Windows Setup validates against a schema, and ONE
bad value invalidates the WHOLE answer file for that pass - the machine stops at
"Windows could not parse or process unattend answer file ... The answer file is
invalid" with nothing configured, and the only clue is a line in
C:\\Windows\\Panther\\setupact.log naming an XPath.

That is not hypothetical. On 2026-08-06 every shopfloor, standard and engineer
build was failing this way:

    /settings/RunSynchronous/RunSynchronousCommand/[Order="16"]/Path
    Description = Value is invalid.   hrResult = 0x80220005   pass = specialize

The cause was an inlined "powershell.exe -Command ..." that had grown to 676
characters in a field capped at 259. It had been broken since the previous
evening. Every check in this script is mechanical and would have caught it before
a single bay was booted.

Checks:
  path-too-long        RunSynchronousCommand/Path        > 259 chars
  cmdline-too-long     SynchronousCommand/CommandLine    > 1024 chars
  description-too-long Description                       > 256 chars
  duplicate-element    an element that may appear once appears twice
  not-well-formed      XML does not parse
  bom                  UTF-8 BOM present (the live files carry none; adding one
                       is an unintended diff, and python's utf-8-sig ADDS one)
  unknown-token        %token% left in the file that PESetup will not substitute
                       (it only replaces %serialnumber% and *arch*)

Usage:
    ./lint-unattend.py                       # every image type on the PXE server
    ./lint-unattend.py --image gea-shopfloor
    ./lint-unattend.py --local playbook/FlatUnattendW10.xml ...
    ./lint-unattend.py --quiet

Exits non-zero when any ERROR is found, so it can gate a deploy.

Remote mode reads PXE_SERVER (default 172.16.9.1), PXE_USER (default pxe)
and PXE_PASS (default pxe) from the environment. The password reaches
sshpass through its environment (sshpass -e), never on a command line.

Requires: sshpass (remote mode only)
"""

import argparse
import os
import re
import shlex
import subprocess
import sys
import xml.dom.minidom

PXE_HOST = os.environ.get("PXE_SERVER") or "172.16.9.1"
PXE_USER = os.environ.get("PXE_USER") or "pxe"
PXE_PASS = os.environ.get("PXE_PASS") or "pxe"
IMAGE_BASE = "/srv/samba/winpeapps"
UNATTEND_REL = "Deploy/FlatUnattendW10.xml"

# Schema limits. Exceeding any of these invalidates the entire answer file for
# the pass the element sits in, not just the one command.
MAX_PATH = 259          # Microsoft-Windows-Deployment RunSynchronousCommand/Path
MAX_COMMANDLINE = 1024  # Microsoft-Windows-Shell-Setup SynchronousCommand/CommandLine
MAX_DESCRIPTION = 256   # Description on either of the above

# PESetup substitutes exactly these two and nothing else (docs/PESETUP-INTERNALS.md).
KNOWN_TOKENS = {"%serialnumber%"}

# Windows expands these itself at run time, so they are not PESetup's problem and
# must not be reported. Without this the check cries wolf on every %WINDIR%.
SHELL_VARS = {"%windir%", "%systemroot%", "%systemdrive%", "%programfiles%",
              "%programdata%", "%temp%", "%tmp%", "%userprofile%", "%appdata%",
              "%localappdata%", "%computername%", "%username%", "%public%",
              "%allusersprofile%", "%path%"}

# Elements that may appear at most once inside a command block.
SINGLE_OCCURRENCE = ("Order", "Description", "Path", "CommandLine", "RequiresUserInput", "WillReboot")

ERROR, WARN = "ERROR", "WARN"


def ssh_cmd(host, cmd):
    env = dict(os.environ, SSHPASS=PXE_PASS)
    return subprocess.run(
        ["sshpass", "-e", "ssh", "-o", "StrictHostKeyChecking=no",
         "-o", "LogLevel=ERROR", f"{PXE_USER}@{host}", cmd],
        capture_output=True, text=True, env=env)


def read_remote_bytes(host, path):
    """Base64 in transit so a BOM or CRLF survives the hop intact."""
    r = ssh_cmd(host, "base64 -w0 %s 2>/dev/null" % shlex.quote(path))
    if r.returncode != 0 or not r.stdout.strip():
        return None
    import base64
    return base64.b64decode(r.stdout.strip())


def read_local_bytes(path):
    try:
        with open(path, "rb") as f:
            return f.read()
    except OSError:
        return None


def blocks(raw, tag):
    return re.findall(r"<%s\b.*?</%s>" % (tag, tag), raw, re.S)


def order_of(block):
    m = re.search(r"<Order>(\d+)</Order>", block)
    return m.group(1) if m else "?"


def check(name, data, findings):
    if data is None:
        findings.append((ERROR, "missing", f"{name}: cannot read the file"))
        return

    if data.startswith(b"\xef\xbb\xbf"):
        findings.append((WARN, "bom",
                         f"{name}: file carries a UTF-8 BOM. The live files have none - "
                         "this is usually an accidental diff from a tool that wrote "
                         "utf-8-sig."))
        raw = data.decode("utf-8-sig")
    else:
        raw = data.decode("utf-8", errors="replace")

    try:
        xml.dom.minidom.parseString(raw.encode("utf-8"))
    except Exception as e:
        findings.append((ERROR, "not-well-formed", f"{name}: XML does not parse: {e}"))
        return

    for tag, field, limit in (("RunSynchronousCommand", "Path", MAX_PATH),
                              ("SynchronousCommand", "CommandLine", MAX_COMMANDLINE)):
        for b in blocks(raw, tag):
            m = re.search(r"<%s>(.*?)</%s>" % (field, field), b, re.S)
            if m and len(m.group(1)) > limit:
                findings.append((ERROR, "%s-too-long" % field.lower(),
                                 f"{name}: {tag} Order {order_of(b)} has {field} of "
                                 f"{len(m.group(1))} chars, limit {limit}. This invalidates "
                                 f"the WHOLE answer file for its pass. Move the body into a "
                                 f"script under Deploy\\Applications and call it by path."))

    for tag in ("RunSynchronousCommand", "SynchronousCommand"):
        for b in blocks(raw, tag):
            for d in re.findall(r"<Description>(.*?)</Description>", b, re.S):
                if len(d) > MAX_DESCRIPTION:
                    findings.append((ERROR, "description-too-long",
                                     f"{name}: {tag} Order {order_of(b)} has a Description of "
                                     f"{len(d)} chars, limit {MAX_DESCRIPTION}. Put the "
                                     f"rationale in an XML comment instead."))
            for el in SINGLE_OCCURRENCE:
                n = len(re.findall(r"<%s>" % el, b))
                if n > 1:
                    findings.append((ERROR, "duplicate-element",
                                     f"{name}: {tag} Order {order_of(b)} has {n} <{el}> "
                                     f"elements; at most one is allowed."))

    for tok in sorted(set(re.findall(r"%[a-zA-Z_]+%", raw))):
        if tok.lower() not in KNOWN_TOKENS and tok.lower() not in SHELL_VARS:
            findings.append((WARN, "unknown-token",
                             f"{name}: {tok} is not substituted by PESetup (it replaces only "
                             f"%serialnumber% and *arch*), so it reaches Windows literally."))


def main():
    parser = argparse.ArgumentParser(description="Validate unattend answer files.")
    parser.add_argument("--server", default=PXE_HOST, help=f"PXE server (default: {PXE_HOST})")
    parser.add_argument("--image", help="single image type instead of all")
    parser.add_argument("--sha256", action="store_true",
                        help="also print the sha256 of each file linted (deploy-unattend.sh compares it)")
    parser.add_argument("--local", nargs="+", metavar="FILE", help="lint local files instead")
    parser.add_argument("--quiet", action="store_true", help="suppress WARN lines")
    args = parser.parse_args()
    if args.image and not re.fullmatch(r"[A-Za-z0-9._-]+", args.image):
        sys.exit("ERROR: bad image type %r" % args.image)

    findings = []
    targets = []

    if args.local:
        for f in args.local:
            targets.append((f, read_local_bytes(f)))
    else:
        if args.image:
            images = [args.image]
        else:
            r = ssh_cmd(args.server,
                        "find %s -maxdepth 1 -mindepth 1 -type d -printf '%%f\\n'" % shlex.quote(IMAGE_BASE))
            if r.returncode != 0:
                sys.exit("ERROR: cannot list %s: %s" % (IMAGE_BASE, r.stderr.strip()))
            images = sorted(n for n in r.stdout.split() if not n.startswith("_"))
        for img in images:
            path = f"{IMAGE_BASE}/{img}/{UNATTEND_REL}"
            targets.append((img, read_remote_bytes(args.server, path)))

    for name, data in targets:
        check(name, data, findings)
        if args.sha256 and data is not None:
            import hashlib
            print("sha256 %s %s" % (hashlib.sha256(data).hexdigest(), name))

    print("Linted %d unattend file(s)\n" % len(targets))
    shown = [f for f in findings if not (args.quiet and f[0] == WARN)]
    for level, kind, msg in sorted(shown, key=lambda f: 0 if f[0] == ERROR else 1):
        print("  %-6s %-22s %s" % (level, kind, msg))
    if not shown:
        print("  clean")

    errors = sum(1 for f in findings if f[0] == ERROR)
    print("\n%d error, %d warn" % (errors, sum(1 for f in findings if f[0] == WARN)))
    if errors:
        print("FAILED: do not deploy this answer file.")
        return 1
    print("PASSED")
    return 0


if __name__ == "__main__":
    sys.exit(main())