f1d9bdf478
A CRLF file re-CRLF'd (e.g. sed adding \r to already-CRLF lines) yields \r\r\n; the old _split_lines left a stray \r that split into a blank line between every line, so parse_boot_menu/settings/lint saw a garbled file and returned nothing. Now collapse any run of CR before a newline. Also re-deployed a clean-CRLF startnet.cmd into the live boot.wim (the earlier sed-based deploys had doubled the CR).
730 lines
26 KiB
Python
730 lines
26 KiB
Python
"""boot.wim manipulation via wimtools (wimextract / wimupdate / wimdir)
|
|
plus pure-Python helpers for the startnet.cmd editor.
|
|
|
|
Two layers live here:
|
|
|
|
1. wimtools wrappers (extract_startnet / update_startnet / list_files) that
|
|
read + write startnet.cmd inside boot.wim. update_startnet writes CRLF.
|
|
|
|
2. Framework-free text helpers used by the rebuilt startnet.cmd editor:
|
|
settings-as-forms (parse_settings / apply_settings), the boot-menu builder
|
|
(parse_boot_menu / apply_boot_menu), lint_startnet, timestamped snapshots
|
|
(save_snapshot / list_snapshots / read_snapshot) and unified_diff.
|
|
|
|
The helpers hold NO Flask imports on purpose; the routes stage wires them up.
|
|
startnet.cmd is a WinPE BATCH script with CRLF endings. Every regen helper
|
|
preserves the file's existing line ending and only rewrites the exact lines it
|
|
targets, leaving everything else byte-for-byte. When a structure cannot be
|
|
located unambiguously the regen helpers raise ValueError so the caller can fall
|
|
back to the raw full-text editor (the source of truth).
|
|
"""
|
|
|
|
import datetime
|
|
import difflib
|
|
import glob
|
|
import os
|
|
import re
|
|
import shutil
|
|
import subprocess
|
|
import tempfile
|
|
from collections import Counter
|
|
|
|
# Root-writable dir for timestamped startnet.cmd snapshots. Overridable by env
|
|
# so tests / alt deploys can redirect it. save_snapshot() creates it on demand.
|
|
BACKUPS_DIR = os.environ.get(
|
|
"STARTNET_BACKUPS_DIR", "/var/lib/pxe-webapp/startnet-backups"
|
|
)
|
|
|
|
|
|
def extract_startnet(wim_path):
|
|
"""Extract startnet.cmd from a WIM file. Returns the contents or None."""
|
|
tmpdir = tempfile.mkdtemp()
|
|
try:
|
|
result = subprocess.run(
|
|
["wimextract", wim_path, "1",
|
|
"/Windows/System32/startnet.cmd",
|
|
"--dest-dir", tmpdir],
|
|
capture_output=True, text=True, timeout=30,
|
|
)
|
|
startnet_path = os.path.join(tmpdir, "startnet.cmd")
|
|
if result.returncode == 0 and os.path.isfile(startnet_path):
|
|
# newline="" keeps the file's CRLF bytes verbatim; default universal
|
|
# newlines would fold CRLF to LF, breaking the lint CRLF check and
|
|
# storing LF-only snapshots.
|
|
with open(startnet_path, "r", encoding="utf-8", errors="replace",
|
|
newline="") as fh:
|
|
return fh.read()
|
|
return None
|
|
except Exception:
|
|
return None
|
|
finally:
|
|
shutil.rmtree(tmpdir, ignore_errors=True)
|
|
|
|
|
|
def update_startnet(wim_path, content):
|
|
"""Update startnet.cmd inside a WIM file via wimupdate.
|
|
|
|
Returns (ok, error_message). Writes CRLF line endings.
|
|
"""
|
|
tmpdir = tempfile.mkdtemp()
|
|
try:
|
|
# Normalize to LF first: newline="\r\n" translates each bare \n, so
|
|
# CRLF input (browser textareas post CRLF) would otherwise become
|
|
# \r\r\n. After normalizing, output is always clean CRLF.
|
|
content = content.replace("\r\n", "\n").replace("\r", "\n")
|
|
startnet_path = os.path.join(tmpdir, "startnet.cmd")
|
|
with open(startnet_path, "w", encoding="utf-8", newline="\r\n") as fh:
|
|
fh.write(content)
|
|
update_cmd = f"add {startnet_path} /Windows/System32/startnet.cmd\n"
|
|
result = subprocess.run(
|
|
["wimupdate", wim_path, "1"],
|
|
input=update_cmd,
|
|
capture_output=True, text=True, timeout=60,
|
|
)
|
|
if result.returncode != 0:
|
|
return False, result.stderr.strip()
|
|
return True, ""
|
|
except Exception as exc:
|
|
return False, str(exc)
|
|
finally:
|
|
shutil.rmtree(tmpdir, ignore_errors=True)
|
|
|
|
|
|
def list_files(wim_path, path="/"):
|
|
"""List files inside a WIM at the given path."""
|
|
try:
|
|
result = subprocess.run(
|
|
["wimdir", wim_path, "1", path],
|
|
capture_output=True, text=True, timeout=30,
|
|
)
|
|
if result.returncode == 0:
|
|
return [l.strip() for l in result.stdout.splitlines() if l.strip()]
|
|
return []
|
|
except Exception:
|
|
return []
|
|
|
|
|
|
# ===========================================================================
|
|
# Pure text helpers for the startnet.cmd editor (no Flask, no wimtools).
|
|
# All operate on the startnet.cmd text. Regen helpers preserve the file's
|
|
# existing line ending and only touch their target lines.
|
|
# ===========================================================================
|
|
|
|
def _detect_eol(content):
|
|
"""Return the dominant line ending, defaulting to CRLF for WinPE batch."""
|
|
if "\r\n" in content:
|
|
return "\r\n"
|
|
if "\n" in content:
|
|
return "\n"
|
|
return "\r\n"
|
|
|
|
|
|
def _split_lines(content):
|
|
"""Split into lines WITHOUT terminators, tolerant of CRLF/LF/CR.
|
|
|
|
Rejoining with eol.join(...) reproduces a trailing newline because a
|
|
trailing terminator yields a final empty element.
|
|
"""
|
|
# Collapse any run of CRs before a newline (handles CRLF and a doubled
|
|
# \r\r\n that a bad CRLF-conversion can leave behind), then lone CR.
|
|
return re.sub(r"\r+\n", "\n", content).replace("\r", "\n").split("\n")
|
|
|
|
|
|
# --- Shared regexes ---------------------------------------------------------
|
|
_LABEL_RE = re.compile(r"^\s*:([A-Za-z0-9_][A-Za-z0-9_\-]*)\s*$")
|
|
_NETUSE_IMG_RE = re.compile(
|
|
r"net use\s+\w+:\s+\\\\[0-9.]+\\winpeapps\\([A-Za-z0-9_\-]+)", re.IGNORECASE
|
|
)
|
|
_GOTO_END_RE = re.compile(r"^\s*goto\s+end\s*$", re.IGNORECASE)
|
|
_DISPATCH_RE = re.compile(
|
|
r'^(\s*)if\s+"%choice%"=="(\d+)"\s+goto\s+([A-Za-z0-9_\-]+)\s*$', re.IGNORECASE
|
|
)
|
|
_ECHO_NUM_RE = re.compile(r"^(\s*)echo\s+(\d+)\.\s?(.*?)\s*$", re.IGNORECASE)
|
|
_CHOICE_PROMPT_RE = re.compile(
|
|
r"^(\s*set\s+/p\s+choice=.*\(1-)(\d+)(\).*)$", re.IGNORECASE
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Feature 1: settings as forms
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _detect_server_ip(content):
|
|
"""Best-effort PXE server IP: the IP most used in \\\\IP\\ UNC paths.
|
|
|
|
Falls back to a `ping <ip>` target. Returns the dotted-quad or None.
|
|
"""
|
|
ips = re.findall(r"\\\\(\d{1,3}(?:\.\d{1,3}){3})\\", content)
|
|
if not ips:
|
|
ips = re.findall(
|
|
r"ping\b[^\r\n]*?\b(\d{1,3}(?:\.\d{1,3}){3})\b", content, re.IGNORECASE
|
|
)
|
|
if not ips:
|
|
return None
|
|
return Counter(ips).most_common(1)[0][0]
|
|
|
|
|
|
def _replace_server_ip(content, old, new):
|
|
"""Replace old IP with new everywhere it stands as a whole IPv4 token.
|
|
|
|
The negative look-around on [\\d.] keeps it from matching a fragment of a
|
|
longer dotted number, and only the detected server IP is ever passed in, so
|
|
unrelated addresses (e.g. 8.8.8.8 in the enroll.cmd here-doc) are untouched.
|
|
This covers UNC (\\\\IP\\...), the ping wait-for-network line, and any URL.
|
|
"""
|
|
pat = re.compile(r"(?<![\d.])" + re.escape(old) + r"(?![\d.])")
|
|
return pat.sub(new, content)
|
|
|
|
|
|
def _detect_menu_timeout_default(content):
|
|
"""Pull timeout seconds + default key off a `choice /t N /d X` line.
|
|
|
|
startnet uses `set /p` (no timeout) today, so this is usually (None, None).
|
|
Only a real `choice` command is recognised; nothing is invented.
|
|
"""
|
|
timeout = None
|
|
default = None
|
|
for line in _split_lines(content):
|
|
if re.search(r"\bchoice\b", line, re.IGNORECASE) and "/t" in line.lower():
|
|
mt = re.search(r"/t\s*:?\s*(\d+)", line, re.IGNORECASE)
|
|
md = re.search(r"/d\s*:?\s*([A-Za-z0-9]+)", line, re.IGNORECASE)
|
|
if mt:
|
|
timeout = int(mt.group(1))
|
|
if md:
|
|
default = md.group(1)
|
|
break
|
|
return timeout, default
|
|
|
|
|
|
def parse_settings(content):
|
|
"""content -> settings dict of safe, common knobs.
|
|
|
|
Keys:
|
|
server_ip str|None PXE server IP used in UNC/ping paths
|
|
menu_timeout int|None seconds, from a `choice /t` construct if present
|
|
menu_default str|None default choice key, from `choice /d` if present
|
|
"""
|
|
ip = _detect_server_ip(content)
|
|
timeout, default = _detect_menu_timeout_default(content)
|
|
return {"server_ip": ip, "menu_timeout": timeout, "menu_default": default}
|
|
|
|
|
|
def apply_settings(content, settings):
|
|
"""settings dict -> content, via targeted replacement only.
|
|
|
|
- server_ip: if given and different from the current detected IP, rewrites
|
|
that IP everywhere it appears as a whole IPv4 token (UNC, ping, URL).
|
|
- menu_timeout / menu_default: updated ONLY if a `choice /t../d..` construct
|
|
already exists; never injected (conservative - avoids risky rewrites).
|
|
Unknown/None keys are left alone. Line endings are preserved.
|
|
"""
|
|
out = content
|
|
new_ip = settings.get("server_ip")
|
|
if new_ip:
|
|
cur = _detect_server_ip(content)
|
|
if cur and cur != new_ip:
|
|
out = _replace_server_ip(out, cur, new_ip)
|
|
|
|
if settings.get("menu_timeout") is not None or settings.get("menu_default") is not None:
|
|
out = _apply_menu_timeout_default(
|
|
out, settings.get("menu_timeout"), settings.get("menu_default")
|
|
)
|
|
return out
|
|
|
|
|
|
def _apply_menu_timeout_default(content, timeout, default):
|
|
"""Rewrite /t and /d on an existing `choice` line only. No-op otherwise."""
|
|
eol = _detect_eol(content)
|
|
lines = _split_lines(content)
|
|
for i, line in enumerate(lines):
|
|
if re.search(r"\bchoice\b", line, re.IGNORECASE) and "/t" in line.lower():
|
|
if timeout is not None:
|
|
line = re.sub(
|
|
r"(/t\s*:?\s*)\d+", r"\g<1>" + str(timeout), line, flags=re.IGNORECASE
|
|
)
|
|
if default is not None:
|
|
line = re.sub(
|
|
r"(/d\s*:?\s*)[A-Za-z0-9]+",
|
|
r"\g<1>" + str(default),
|
|
line,
|
|
flags=re.IGNORECASE,
|
|
)
|
|
lines[i] = line
|
|
break
|
|
return eol.join(lines)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Feature 2: boot-menu builder
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _find_label_line(lines, name):
|
|
name = name.lower()
|
|
for i, line in enumerate(lines):
|
|
m = _LABEL_RE.match(line)
|
|
if m and m.group(1).lower() == name:
|
|
return i
|
|
return None
|
|
|
|
|
|
def _find_image_blocks(lines):
|
|
"""Find per-image action blocks: a ':<target>' label whose body maps
|
|
Z: to \\\\IP\\winpeapps\\<image> and ends with 'goto end'.
|
|
|
|
Returns list of dicts: {target, image, start, end} where start is the label
|
|
line index and end is the 'goto end' line index (inclusive block span).
|
|
"""
|
|
label_idxs = [i for i, l in enumerate(lines) if _LABEL_RE.match(l)]
|
|
label_set = set(label_idxs)
|
|
blocks = []
|
|
for i in label_idxs:
|
|
target = _LABEL_RE.match(lines[i]).group(1)
|
|
# body runs to the next label or EOF
|
|
end_bound = len(lines)
|
|
for j in range(i + 1, len(lines)):
|
|
if j in label_set:
|
|
end_bound = j
|
|
break
|
|
image = None
|
|
goto_end_idx = None
|
|
for j in range(i + 1, end_bound):
|
|
if image is None:
|
|
mi = _NETUSE_IMG_RE.search(lines[j])
|
|
if mi:
|
|
image = mi.group(1)
|
|
if _GOTO_END_RE.match(lines[j]):
|
|
goto_end_idx = j
|
|
if image is not None and goto_end_idx is not None:
|
|
blocks.append(
|
|
{"target": target, "image": image, "start": i, "end": goto_end_idx}
|
|
)
|
|
return blocks
|
|
|
|
|
|
def _find_dispatch_lines(lines):
|
|
"""All `if "%choice%"=="N" goto TARGET` lines as dicts {idx,num,target,indent}."""
|
|
out = []
|
|
for i, line in enumerate(lines):
|
|
m = _DISPATCH_RE.match(line)
|
|
if m:
|
|
out.append(
|
|
{
|
|
"idx": i,
|
|
"indent": m.group(1),
|
|
"num": int(m.group(2)),
|
|
"target": m.group(3),
|
|
}
|
|
)
|
|
return out
|
|
|
|
|
|
def _find_choice_prompt(lines, start):
|
|
for i in range(start, len(lines)):
|
|
if _CHOICE_PROMPT_RE.match(lines[i]):
|
|
return i
|
|
return None
|
|
|
|
|
|
def _find_numbered_echo_run(lines, lo, hi):
|
|
"""Contiguous run of numbered 'echo N. Label' lines within [lo, hi).
|
|
|
|
Returns (indices, indent, labels_by_num) or raises ValueError if the numbered
|
|
echo lines are not on consecutive source lines (ambiguous - refuse to edit).
|
|
"""
|
|
matched = []
|
|
for i in range(lo, hi):
|
|
m = _ECHO_NUM_RE.match(lines[i])
|
|
if m:
|
|
matched.append((i, m))
|
|
if not matched:
|
|
raise ValueError("no numbered 'echo N. ...' menu lines found")
|
|
idxs = [i for i, _ in matched]
|
|
if idxs != list(range(idxs[0], idxs[0] + len(idxs))):
|
|
raise ValueError("menu echo lines are not contiguous")
|
|
indent = matched[0][1].group(1)
|
|
labels = {int(m.group(2)): m.group(3).rstrip() for _, m in matched}
|
|
return idxs, indent, labels
|
|
|
|
|
|
def _analyze_boot_menu(lines):
|
|
"""Locate every editable region of the image menu. Raises ValueError if the
|
|
clearly-delimited structure is missing/ambiguous. Returns a dict of parts."""
|
|
menu_idx = _find_label_line(lines, "menu")
|
|
if menu_idx is None:
|
|
raise ValueError("no :menu label found")
|
|
choice_idx = _find_choice_prompt(lines, menu_idx)
|
|
if choice_idx is None:
|
|
raise ValueError("no 'set /p choice=' prompt found after :menu")
|
|
|
|
echo_idxs, echo_indent, echo_labels = _find_numbered_echo_run(
|
|
lines, menu_idx, choice_idx
|
|
)
|
|
|
|
image_blocks = _find_image_blocks(lines)
|
|
if not image_blocks:
|
|
raise ValueError("no image action blocks (net use winpeapps ... goto end)")
|
|
img_targets = {b["target"] for b in image_blocks}
|
|
|
|
all_dispatch = _find_dispatch_lines(lines)
|
|
image_dispatch = [d for d in all_dispatch if d["target"] in img_targets]
|
|
if not image_dispatch:
|
|
raise ValueError("no image %choice% dispatch lines found")
|
|
d_idxs = [d["idx"] for d in image_dispatch]
|
|
if d_idxs != list(range(d_idxs[0], d_idxs[-1] + 1)):
|
|
raise ValueError("image %choice% dispatch lines are not contiguous")
|
|
|
|
blocks_sorted = sorted(image_blocks, key=lambda b: b["start"])
|
|
region_start = blocks_sorted[0]["start"]
|
|
region_end = blocks_sorted[-1]["end"]
|
|
# every line inside the block region must belong to a block span or be blank
|
|
covered = set()
|
|
for b in blocks_sorted:
|
|
covered.update(range(b["start"], b["end"] + 1))
|
|
for j in range(region_start, region_end + 1):
|
|
if j not in covered and lines[j].strip() != "":
|
|
raise ValueError(
|
|
"image action blocks are not contiguous (foreign content at "
|
|
"line %d)" % (j + 1)
|
|
)
|
|
|
|
return {
|
|
"menu_idx": menu_idx,
|
|
"choice_idx": choice_idx,
|
|
"echo_idxs": echo_idxs,
|
|
"echo_indent": echo_indent,
|
|
"echo_labels": echo_labels,
|
|
"image_blocks": blocks_sorted,
|
|
"image_dispatch": image_dispatch,
|
|
"all_dispatch": all_dispatch,
|
|
"region_start": region_start,
|
|
"region_end": region_end,
|
|
"dispatch_start": d_idxs[0],
|
|
"dispatch_end": d_idxs[-1],
|
|
}
|
|
|
|
|
|
def parse_boot_menu(content):
|
|
"""content -> list of image-menu items, ordered by menu number.
|
|
|
|
Each item dict:
|
|
num int choice number that routes to this image
|
|
label str echo text after 'N. ' (kept verbatim, trailing ws only stripped)
|
|
target str goto label of the action block (the ':<target>')
|
|
target_image str winpeapps\\<image> the block maps Z: to
|
|
|
|
Returns [] if the delimited menu region cannot be found (caller uses raw text).
|
|
"""
|
|
lines = _split_lines(content)
|
|
try:
|
|
a = _analyze_boot_menu(lines)
|
|
except ValueError:
|
|
return []
|
|
by_target = {b["target"]: b for b in a["image_blocks"]}
|
|
items = []
|
|
for d in a["image_dispatch"]:
|
|
blk = by_target[d["target"]]
|
|
items.append(
|
|
{
|
|
"num": d["num"],
|
|
"label": a["echo_labels"].get(d["num"], ""),
|
|
"target": d["target"],
|
|
"target_image": blk["image"],
|
|
}
|
|
)
|
|
items.sort(key=lambda it: it["num"])
|
|
return items
|
|
|
|
|
|
def _synth_block(item, ip, eol_indent=""):
|
|
"""Build a fresh per-image action block for a newly added menu item."""
|
|
target = item["target"]
|
|
image = item.get("target_image") or target
|
|
label = item.get("label") or target
|
|
return [
|
|
":%s" % target,
|
|
"echo.",
|
|
"echo Starting %s setup..." % label,
|
|
r'start "FlatApp" %SYSTEMDRIVE%\GESetup\FlatSetupLoader.exe',
|
|
r"for /l %%i in (1,1,2000000) do rem",
|
|
r"net use Z: \\%s\winpeapps\%s /user:pxe-upload pxe /persistent:no"
|
|
% (ip, image),
|
|
"goto end",
|
|
]
|
|
|
|
|
|
def apply_boot_menu(content, items):
|
|
"""items -> content: regenerate ONLY the menu echo lines, the choice prompt
|
|
count, the image %choice% dispatch (renumbered 1..N), and reorder/add/remove
|
|
the per-image action blocks. Existing block BODIES are kept byte-for-byte;
|
|
only new blocks are synthesised. Everything else is untouched.
|
|
|
|
items: list of {label, target, target_image} (a 'num' key, if present, is
|
|
ignored - order in the list defines the new numbering).
|
|
|
|
Raises ValueError when the structure is ambiguous OR when renumbering would
|
|
desync a secondary `%choice%` router (e.g. the enrollment routing that also
|
|
keys off choice numbers). The caller then falls back to the raw editor.
|
|
"""
|
|
if not items:
|
|
raise ValueError("boot menu must have at least one item")
|
|
|
|
eol = _detect_eol(content)
|
|
lines = _split_lines(content)
|
|
ip = _detect_server_ip(content) or "172.16.9.1"
|
|
a = _analyze_boot_menu(lines)
|
|
|
|
old_num_by_target = {d["target"]: d["num"] for d in a["image_dispatch"]}
|
|
old_target_by_num = {d["num"]: d["target"] for d in a["image_dispatch"]}
|
|
new_num_by_target = {it["target"]: i + 1 for i, it in enumerate(items)}
|
|
|
|
# Safety: a secondary %choice% router (not one of the image blocks) that
|
|
# references an image number whose meaning is about to change is a desync.
|
|
image_dispatch_idxs = {d["idx"] for d in a["image_dispatch"]}
|
|
for d in a["all_dispatch"]:
|
|
if d["idx"] in image_dispatch_idxs:
|
|
continue
|
|
n = d["num"]
|
|
if n in old_target_by_num: # this number currently selects an image
|
|
tgt = old_target_by_num[n]
|
|
if tgt not in new_num_by_target or new_num_by_target[tgt] != n:
|
|
raise ValueError(
|
|
"refusing to renumber: secondary %%choice%% routing at line "
|
|
"%d references image choice %d whose number would change"
|
|
% (d["idx"] + 1, n)
|
|
)
|
|
|
|
blocks_by_target = {
|
|
b["target"]: lines[b["start"]:b["end"] + 1] for b in a["image_blocks"]
|
|
}
|
|
|
|
# --- build replacement lines for each region ---
|
|
new_echo = [
|
|
"%secho %d. %s" % (a["echo_indent"], i + 1, it.get("label") or it["target"])
|
|
for i, it in enumerate(items)
|
|
]
|
|
disp_indent = a["image_dispatch"][0]["indent"]
|
|
new_dispatch = [
|
|
'%sif "%%choice%%"=="%d" goto %s' % (disp_indent, i + 1, it["target"])
|
|
for i, it in enumerate(items)
|
|
]
|
|
cm = _CHOICE_PROMPT_RE.match(lines[a["choice_idx"]])
|
|
new_choice = "%s%d%s" % (cm.group(1), len(items), cm.group(3))
|
|
|
|
new_region = []
|
|
for i, it in enumerate(items):
|
|
if i > 0:
|
|
new_region.append("") # blank line between blocks, matching style
|
|
if it["target"] in blocks_by_target:
|
|
new_region.extend(blocks_by_target[it["target"]])
|
|
else:
|
|
new_region.extend(_synth_block(it, ip))
|
|
|
|
# Apply splices high-index-first so earlier indices stay valid.
|
|
splices = [
|
|
(a["region_start"], a["region_end"], new_region),
|
|
(a["dispatch_start"], a["dispatch_end"], new_dispatch),
|
|
(a["choice_idx"], a["choice_idx"], [new_choice]),
|
|
(a["echo_idxs"][0], a["echo_idxs"][-1], new_echo),
|
|
]
|
|
for start, end, repl in sorted(splices, key=lambda s: s[0], reverse=True):
|
|
lines[start:end + 1] = repl
|
|
|
|
return eol.join(lines)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Feature 3: lint
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _strip_for_parens(line):
|
|
"""Neutralise a line for paren counting: drop comments, quoted spans, and
|
|
caret-escaped chars (so `^>NUL`, `echo ... ^(` etc. do not count)."""
|
|
s = line.lstrip()
|
|
low = s.lower()
|
|
if low.startswith("rem ") or low == "rem" or s.startswith("::"):
|
|
return ""
|
|
# remove escaped pairs first, then any caret-escaped char
|
|
s = s.replace("^^", "")
|
|
s = re.sub(r"\^.", "", s)
|
|
# drop double-quoted spans
|
|
s = re.sub(r'"[^"]*"', "", s)
|
|
return s
|
|
|
|
|
|
def lint_startnet(content):
|
|
"""Static checks over startnet.cmd. Advisory - never blocks a save.
|
|
|
|
Returns {"errors": [...], "warnings": [...]} where each entry is
|
|
{"line": int|None, "message": str}. Errors: goto with no matching label.
|
|
Warnings: lone-LF (non-CRLF) lines, unbalanced parentheses.
|
|
"""
|
|
errors = []
|
|
warnings = []
|
|
lines = _split_lines(content)
|
|
|
|
labels = set()
|
|
for line in lines:
|
|
m = _LABEL_RE.match(line)
|
|
if m:
|
|
labels.add(m.group(1).lower())
|
|
|
|
for i, line in enumerate(lines):
|
|
# A `goto` inside an `echo` line is emitted text (e.g. the enroll.cmd
|
|
# here-doc), not a jump - skip it to avoid false positives.
|
|
if line.lstrip().lower().startswith("echo"):
|
|
continue
|
|
for m in re.finditer(
|
|
r"goto\s+:?([A-Za-z0-9_][A-Za-z0-9_\-]*)", line, re.IGNORECASE
|
|
):
|
|
target = m.group(1).lower()
|
|
if target == "eof":
|
|
continue # goto :eof is a cmd.exe builtin
|
|
if target not in labels:
|
|
errors.append(
|
|
{
|
|
"line": i + 1,
|
|
"message": "goto %s has no matching :%s label"
|
|
% (m.group(1), m.group(1)),
|
|
}
|
|
)
|
|
|
|
# CRLF check: count lone LFs (a \n not preceded by \r)
|
|
crlf = content.count("\r\n")
|
|
lf = content.count("\n")
|
|
lone = lf - crlf
|
|
if lone > 0:
|
|
first = None
|
|
idx = content.find("\n")
|
|
ln = 1
|
|
while idx != -1:
|
|
if idx == 0 or content[idx - 1] != "\r":
|
|
first = ln
|
|
break
|
|
idx = content.find("\n", idx + 1)
|
|
ln += 1
|
|
warnings.append(
|
|
{
|
|
"line": first,
|
|
"message": "%d line(s) are not CRLF-terminated; WinPE batch "
|
|
"expects CRLF" % lone,
|
|
}
|
|
)
|
|
|
|
# Parenthesis balance (best-effort, comments/quotes/carets removed)
|
|
delta = 0
|
|
for line in lines:
|
|
cleaned = _strip_for_parens(line)
|
|
delta += cleaned.count("(") - cleaned.count(")")
|
|
if delta != 0:
|
|
warnings.append(
|
|
{
|
|
"line": None,
|
|
"message": "parentheses look unbalanced (delta %+d) - check "
|
|
"if(...) blocks" % delta,
|
|
}
|
|
)
|
|
|
|
return {"errors": errors, "warnings": warnings}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Feature 4: snapshots + diff
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _ensure_backups_dir():
|
|
os.makedirs(BACKUPS_DIR, exist_ok=True)
|
|
|
|
|
|
def _snapshot_meta(path):
|
|
st = os.stat(path)
|
|
base = os.path.basename(path)
|
|
note = ""
|
|
note_path = path + ".note"
|
|
if os.path.isfile(note_path):
|
|
try:
|
|
with open(note_path, "r", encoding="utf-8", errors="replace") as fh:
|
|
note = fh.read().strip()
|
|
except Exception:
|
|
note = ""
|
|
return {
|
|
"id": base,
|
|
"path": path,
|
|
"mtime": st.st_mtime,
|
|
"timestamp": datetime.datetime.fromtimestamp(st.st_mtime).strftime(
|
|
"%Y-%m-%d %H:%M:%S"
|
|
),
|
|
"size": st.st_size,
|
|
"note": note,
|
|
}
|
|
|
|
|
|
def save_snapshot(content, note=""):
|
|
"""Write content verbatim to a timestamped file under BACKUPS_DIR.
|
|
|
|
Snapshots the bytes exactly (no line-ending translation) so a restore
|
|
round-trips faithfully. Returns the snapshot metadata dict (see
|
|
_snapshot_meta). Call this with the PRIOR startnet before overwriting it.
|
|
"""
|
|
_ensure_backups_dir()
|
|
ts = datetime.datetime.now().strftime("%Y%m%d-%H%M%S")
|
|
base = "startnet-%s.cmd" % ts
|
|
path = os.path.join(BACKUPS_DIR, base)
|
|
n = 1
|
|
while os.path.exists(path):
|
|
base = "startnet-%s-%d.cmd" % (ts, n)
|
|
path = os.path.join(BACKUPS_DIR, base)
|
|
n += 1
|
|
with open(path, "w", encoding="utf-8", newline="") as fh:
|
|
fh.write(content)
|
|
if note:
|
|
with open(path + ".note", "w", encoding="utf-8", newline="") as fh:
|
|
fh.write(note)
|
|
return _snapshot_meta(path)
|
|
|
|
|
|
def list_snapshots():
|
|
"""Return snapshot metadata dicts, newest first."""
|
|
_ensure_backups_dir()
|
|
metas = []
|
|
for path in glob.glob(os.path.join(BACKUPS_DIR, "startnet-*.cmd")):
|
|
if path.endswith(".note"):
|
|
continue
|
|
try:
|
|
metas.append(_snapshot_meta(path))
|
|
except OSError:
|
|
continue
|
|
metas.sort(key=lambda m: m["mtime"], reverse=True)
|
|
return metas
|
|
|
|
|
|
def read_snapshot(snapshot_id):
|
|
"""Return a snapshot's text by id (its filename), or None if not found.
|
|
|
|
Rejects path traversal: the id must be a bare basename under BACKUPS_DIR.
|
|
"""
|
|
if not snapshot_id or os.path.basename(snapshot_id) != snapshot_id:
|
|
return None
|
|
if ".." in snapshot_id or "/" in snapshot_id or "\\" in snapshot_id:
|
|
return None
|
|
path = os.path.join(BACKUPS_DIR, snapshot_id)
|
|
if not os.path.isfile(path):
|
|
return None
|
|
# newline="" so the stored CRLF bytes come back verbatim (faithful restore).
|
|
with open(path, "r", encoding="utf-8", errors="replace", newline="") as fh:
|
|
return fh.read()
|
|
|
|
|
|
def unified_diff(a, b, fromfile="current", tofile="new"):
|
|
"""Unified diff between two startnet texts, as a single string ('' if equal).
|
|
|
|
Line endings are normalised for comparison so a pure CRLF/LF difference does
|
|
not produce a wall of noise; content differences are what surface.
|
|
"""
|
|
a_lines = _split_lines(a)
|
|
b_lines = _split_lines(b)
|
|
return "\n".join(
|
|
difflib.unified_diff(
|
|
a_lines, b_lines, fromfile=fromfile, tofile=tofile, lineterm=""
|
|
)
|
|
)
|