ge-aerospace/pxe-server
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Fix PXE interface detection, add br-pxe bridge to test VM, network upload import

Browse Source 
- Playbook: detect interface already configured with 10.9.100.1 before
  falling back to non-default-gateway heuristic (fixes dnsmasq binding
  to wrong NIC when multiple interfaces exist)
- test-vm.sh: auto-attach br-pxe bridge NIC if available on host
- Webapp: add network upload import via SMB share with shared driver
  deduplication and symlinks

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
cproudlock
2026-02-11 15:15:14 -05:00
parent 7486b9ed66
commit f4c158a5ac
4 changed files with 290 additions and 37 deletions
Download Patch File Download Diff File Expand all files Collapse all files

183
webapp/app.py
View File

@@ -51,6 +51,20 @@ def audit(action, detail=""):
SAMBA_SHARE = os.environ.get("SAMBA_SHARE", "/srv/samba/winpeapps")
CLONEZILLA_SHARE = os.environ.get("CLONEZILLA_SHARE", "/srv/samba/clonezilla")
BLANCCO_REPORTS = os.environ.get("BLANCCO_REPORTS", "/srv/samba/blancco-reports")
UPLOAD_DIR = os.environ.get("UPLOAD_DIR", "/home/pxe/image-upload")
SHARED_DIR = os.path.join(SAMBA_SHARE, "_shared")
# Subdirs inside Deploy/ shared across ALL image types
SHARED_DEPLOY_GLOBAL = ["Out-of-box Drivers"]
# Subdirs inside Deploy/ shared within the same image family (by prefix)
SHARED_DEPLOY_SCOPED = {
"gea-": ["Operating Systems"],
"ge-": ["Operating Systems"],
}
# Sibling dirs at image root shared within the same image family
SHARED_ROOT_DIRS = {
"gea-": ["Sources"],
"ge-": ["Sources"],
}
WEB_ROOT = os.environ.get("WEB_ROOT", "/var/www/html")
BOOT_WIM = os.path.join(WEB_ROOT, "win11", "sources", "boot.wim")
@@ -118,6 +132,11 @@ def qwcm(attr):
# Utility helpers
# ---------------------------------------------------------------------------
def image_root(image_type):
"""Return the root directory for an image type."""
return os.path.join(SAMBA_SHARE, image_type)
def deploy_path(image_type):
"""Return the Deploy directory for an image type."""
return os.path.join(SAMBA_SHARE, image_type, "Deploy")
@@ -175,6 +194,103 @@ def find_usb_mounts():
return sorted(set(mounts))
def find_upload_sources():
"""Return sub-directories inside UPLOAD_DIR that look like image content."""
sources = []
if os.path.isdir(UPLOAD_DIR):
# Include the upload dir itself if it has content
try:
entries = os.listdir(UPLOAD_DIR)
if entries:
sources.append(UPLOAD_DIR)
# Also include immediate subdirectories
for entry in entries:
full = os.path.join(UPLOAD_DIR, entry)
if os.path.isdir(full):
sources.append(full)
except OSError:
pass
return sources
def _import_deploy(src_deploy, dst_deploy, target=""):
"""Copy Deploy directory contents, merging shared subdirs into _shared."""
# Build list of scoped shared dirs for this target
scoped_shared = []
prefix_key = ""
for prefix, dirs in SHARED_DEPLOY_SCOPED.items():
if target.startswith(prefix):
scoped_shared = dirs
prefix_key = prefix
break
os.makedirs(dst_deploy, exist_ok=True)
for item in os.listdir(src_deploy):
src_item = os.path.join(src_deploy, item)
dst_item = os.path.join(dst_deploy, item)
if not os.path.isdir(src_item):
shutil.copy2(src_item, dst_item)
continue
# Global shared (e.g., Out-of-box Drivers) — one copy for all
if item in SHARED_DEPLOY_GLOBAL:
shared_dest = os.path.join(SHARED_DIR, item)
os.makedirs(shared_dest, exist_ok=True)
_merge_tree(src_item, shared_dest)
_replace_with_symlink(dst_item, shared_dest)
continue
# Scoped shared (e.g., Operating Systems) — per family prefix
if item in scoped_shared:
shared_dest = os.path.join(SHARED_DIR, f"{prefix_key}{item}")
os.makedirs(shared_dest, exist_ok=True)
_merge_tree(src_item, shared_dest)
_replace_with_symlink(dst_item, shared_dest)
continue
# Normal copy
if os.path.exists(dst_item):
shutil.rmtree(dst_item)
shutil.copytree(src_item, dst_item)
def _replace_with_symlink(link_path, target_path):
"""Replace a file/dir/symlink at link_path with a symlink to target_path."""
if os.path.islink(link_path):
os.remove(link_path)
elif os.path.isdir(link_path):
shutil.rmtree(link_path)
os.symlink(target_path, link_path)
def _merge_tree(src, dst):
"""Recursively merge src tree into dst, overwriting existing files."""
for item in os.listdir(src):
s = os.path.join(src, item)
d = os.path.join(dst, item)
if os.path.isdir(s):
if os.path.isdir(d):
_merge_tree(s, d)
else:
if os.path.exists(d):
os.remove(d)
shutil.copytree(s, d)
else:
os.makedirs(os.path.dirname(d), exist_ok=True)
shutil.copy2(s, d)
def allowed_import_source(source):
"""Check if a source path is a valid import location (USB or upload dir)."""
usb = find_usb_mounts()
if any(source == m or source.startswith(m + "/") for m in usb):
return True
if source == UPLOAD_DIR or source.startswith(UPLOAD_DIR + "/"):
return os.path.isdir(source)
return False
# ---------------------------------------------------------------------------
# XML helpers — parse / build unattend.xml
# ---------------------------------------------------------------------------
@@ -514,6 +630,7 @@ def dashboard():
@app.route("/images/import", methods=["GET", "POST"])
def images_import():
usb_mounts = find_usb_mounts()
upload_sources = find_upload_sources()
images = [image_status(it) for it in IMAGE_TYPES]
if request.method == "POST":
@@ -528,32 +645,67 @@ def images_import():
flash("Invalid target image type.", "danger")
return redirect(url_for("images_import"))
# Validate source is under an allowed USB mount path
allowed = find_usb_mounts()
if not any(source == m or source.startswith(m + "/") for m in allowed):
flash("Source path is not on a mounted USB device.", "danger")
if not 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 = image_root(target)
dest = deploy_path(target)
try:
os.makedirs(dest, exist_ok=True)
# Use rsync-style copy: copy contents of source into dest
for item in os.listdir(source):
src_item = os.path.join(source, item)
dst_item = os.path.join(dest, item)
if os.path.isdir(src_item):
if os.path.exists(dst_item):
shutil.rmtree(dst_item)
shutil.copytree(src_item, dst_item)
else:
shutil.copy2(src_item, dst_item)
src_items = os.listdir(source)
# Detect layout: if source has Deploy/, Sources/, Tools/ at top
# level, it's the full image root structure (USB-style).
# Otherwise treat it as Deploy/ contents directly.
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:
# Determine which root-level dirs are shared for this target
shared_root = []
for prefix, dirs in SHARED_ROOT_DIRS.items():
if target.startswith(prefix):
shared_root = dirs
break
# Full image root: copy Deploy contents + sibling dirs
for item in src_items:
src_item = os.path.join(source, item)
if item == "Deploy":
_import_deploy(src_item, dest, target)
elif os.path.isdir(src_item) and item in shared_root:
# Shared sibling: merge into _shared/{prefix}{item}
# and symlink from image root
prefix_key = target.split("-")[0] + "-"
shared_dest = os.path.join(SHARED_DIR, f"{prefix_key}{item}")
os.makedirs(shared_dest, exist_ok=True)
_merge_tree(src_item, shared_dest)
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):
# Non-shared sibling dirs (Tools) go into image root
dst_item = os.path.join(root, item)
if os.path.exists(dst_item):
shutil.rmtree(dst_item)
shutil.copytree(src_item, dst_item)
else:
shutil.copy2(src_item, os.path.join(root, item))
else:
# Flat layout: treat source as Deploy contents
_import_deploy(source, dest, target)
audit("IMAGE_IMPORT", f"{source} -> {target}")
flash(
f"Successfully imported content from {source} to {FRIENDLY_NAMES.get(target, target)}.",
f"Successfully imported content to {FRIENDLY_NAMES.get(target, target)}.",
"success",
)
except Exception as exc:
@@ -564,6 +716,7 @@ def images_import():
return render_template(
"import.html",
usb_mounts=usb_mounts,
upload_sources=upload_sources,
images=images,
image_types=IMAGE_TYPES,
friendly_names=FRIENDLY_NAMES,