Aria-Software/macos-usb-stick-creator
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Initial commit: macOS USB Stick Creator - Docker Compose Web-GUI fuer bootfaehige macOS-Installer

Browse Source
This commit is contained in:
ARIA
2026-07-18 08:25:03 +00:00
commit e93ab96faa
11 changed files with 1037 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/app.py
+117
View File
@@ -0,0 +1,117 @@
import json
import logging
import os
import threading
import time
from flask import Flask, Response, jsonify, request, send_from_directory
import catalog
import installer
from devices import list_usb_sticks
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(name)s: %(message)s")
log = logging.getLogger("app")
FRONTEND_DIR = os.path.join(os.path.dirname(__file__), "..", "frontend")
app = Flask(__name__, static_folder=None)
_catalog_cache = {"items": [], "catalog_url": None, "loaded_at": None, "error": None, "loading": True}
_catalog_lock = threading.Lock()
def _load_catalog():
with _catalog_lock:
_catalog_cache["loading"] = True
try:
items, url = catalog.list_macos_installers()
with _catalog_lock:
_catalog_cache.update(
items=items, catalog_url=url, loaded_at=time.time(), error=None, loading=False
)
log.info("Katalog geladen: %d macOS-Installer gefunden (%s)", len(items), url)
except Exception as exc: # noqa: BLE001
log.exception("Katalog-Laden fehlgeschlagen")
with _catalog_lock:
_catalog_cache.update(error=str(exc), loading=False)
# Beim Container-Start einmal im Hintergrund laden, damit die GUI sofort
# ansprechbar ist und nicht auf den ersten Request wartet.
threading.Thread(target=_load_catalog, daemon=True).start()
@app.get("/")
def index():
return send_from_directory(FRONTEND_DIR, "index.html")
@app.get("/<path:filename>")
def static_files(filename):
return send_from_directory(FRONTEND_DIR, filename)
@app.get("/api/versions")
def api_versions():
with _catalog_lock:
return jsonify(dict(_catalog_cache))
@app.post("/api/versions/refresh")
def api_versions_refresh():
threading.Thread(target=_load_catalog, daemon=True).start()
return jsonify({"ok": True})
@app.get("/api/devices")
def api_devices():
try:
return jsonify({"items": list_usb_sticks()})
except Exception as exc: # noqa: BLE001
log.exception("device listing failed")
return jsonify({"items": [], "error": str(exc)}), 500
@app.post("/api/jobs")
def api_create_job():
body = request.get_json(force=True)
product_id = body.get("product_id")
device_path = body.get("device_path")
confirm_path = body.get("confirm_device_path")
with _catalog_lock:
product = next((p for p in _catalog_cache["items"] if p["id"] == product_id), None)
if not product:
return jsonify({"error": "Unbekannte macOS-Version -- Katalog neu laden."}), 400
try:
job = installer.create_job(product, device_path, confirm_path)
except ValueError as exc:
return jsonify({"error": str(exc)}), 400
return jsonify({"job_id": job.id})
@app.get("/api/jobs/<job_id>")
def api_job_status(job_id):
job = installer.JOBS.get(job_id)
if not job:
return jsonify({"error": "unbekannter Job"}), 404
return jsonify(job.snapshot())
@app.get("/api/jobs/<job_id>/stream")
def api_job_stream(job_id):
job = installer.JOBS.get(job_id)
if not job:
return jsonify({"error": "unbekannter Job"}), 404
def gen():
for event in job.stream():
yield f"data: {json.dumps(event)}\n\n"
return Response(gen(), mimetype="text/event-stream")
if __name__ == "__main__":
app.run(host="0.0.0.0", port=8080, threaded=True)
backend/catalog.py
+158
View File
@@ -0,0 +1,158 @@
"""
Apple Software-Update-Katalog: Liste verfuegbarer macOS-Installer laden.
Funktionsweise (live gegen swscan.apple.com verifiziert, 2026-07-18):
Apple veroeffentlicht einen kumulativen Produktkatalog unter einer URL, deren
Dateiname die Kette der unterstuetzten Major-Versionen absteigend auflistet,
z.B. fuer den aktuell (Stand Verifikation) neuesten Stand:
index-16-15-14-13-12-10.16-10.15-10.14-10.13-10.12-10.11-10.10-10.9-
mountainlion-lion-snowleopard-leopard.merged-1.sucatalog.gz
Jedes Mal wenn Apple eine neue macOS-Hauptversion (naechstes Jahr z.B. 17)
veroeffentlicht, legt Apple einen NEUEN Katalog mit einer um 1 hoeheren
fuehrenden Zahl an ("index-17-16-15-..."). Der neue Katalog ist kumulativ --
er enthaelt weiterhin alle aelteren Installer. Wir muessen also nur bei jedem
Start herausfinden, welche fuehrende Zahl AKTUELL die hoechste gueltige ist --
das macht find_latest_catalog_url() per HTTP-Probing.
WICHTIG / Grenze der Zukunftssicherheit: Das Namensschema selbst
(("index-<N>-<N-1>-...-legacy-suffix.sucatalog.gz") liegt in Apples Hand.
Es ist seit vielen Jahren stabil, aber falls Apple es grundlegend aendert,
muss dieses Modul angepasst werden -- 100% garantiert "nie wieder anfassen"
gibt es bei einer nicht dokumentierten Drittanbieter-Schnittstelle nicht.
Was wir bieten: neue macOS-Versionen tauchen OHNE Code-Aenderung automatisch
auf, solange Apple das Schema beibehaelt.
"""
import gzip
import logging
import plistlib
import requests
log = logging.getLogger("catalog")
CATALOG_HOST = "https://swscan.apple.com/content/catalogs/others/"
LEGACY_SUFFIX = (
"10.16-10.15-10.14-10.13-10.12-10.11-10.10-10.9-"
"mountainlion-lion-snowleopard-leopard.merged-1.sucatalog.gz"
)
FLOOR = 12 # aelteste bekannte Kette, ab der wir hochzaehlen
PROBE_HEADROOM = 15 # so viele Fehlschlaege in Folge tolerieren wir, bevor wir abbrechen
def _build_catalog_url(max_major: int) -> str:
chain = "-".join(str(n) for n in range(max_major, 11, -1))
return f"{CATALOG_HOST}index-{chain}-{LEGACY_SUFFIX}"
def find_latest_catalog_url(timeout: float = 8) -> str:
best = None
misses = 0
m = FLOOR
while misses < PROBE_HEADROOM:
url = _build_catalog_url(m)
try:
r = requests.head(url, timeout=timeout)
ok = r.status_code == 200
except requests.RequestException:
ok = False
if ok:
best = url
misses = 0
else:
misses += 1
m += 1
if not best:
raise RuntimeError(
"Konnte keinen Apple SUCatalog finden -- Apple hat vermutlich das "
"URL-Schema geaendert. catalog.py muss aktualisiert werden."
)
return best
def _fetch_catalog_products(url: str, timeout: float = 60) -> dict:
r = requests.get(url, timeout=timeout)
r.raise_for_status()
raw = gzip.decompress(r.content)
data = plistlib.loads(raw)
return data.get("Products", {})
def _is_macos_installer(product: dict) -> bool:
emi = product.get("ExtendedMetaInfo", {}) or {}
ids = emi.get("InstallAssistantPackageIdentifiers", {}) or {}
return ids.get("OSInstall") == "com.apple.mpkg.OSInstall"
def _fetch_product_metadata(product: dict, timeout: float = 20):
smd_url = product.get("ServerMetadataURL")
title, version = None, None
if smd_url:
try:
r = requests.get(smd_url, timeout=timeout)
r.raise_for_status()
meta = plistlib.loads(r.content)
version = meta.get("CFBundleShortVersionString")
loc = meta.get("localization", {}) or {}
en = loc.get("English") or (next(iter(loc.values())) if loc else {})
title = en.get("title")
except Exception as exc: # noqa: BLE001
log.warning("Metadaten-Fetch fehlgeschlagen fuer %s: %s", smd_url, exc)
return title, version
def _find_packages(product: dict):
"""Liefert (install_assistant_pkg, base_system_dmg, total_size_bytes)."""
packages = product.get("Packages", []) or []
install_assistant = None
base_system = None
total_size = 0
for p in packages:
url = p.get("URL", "")
size = p.get("Size", 0) or 0
total_size += size
if url.endswith("InstallAssistant.pkg"):
install_assistant = {"url": url, "size": size}
if url.endswith("BaseSystem.dmg") or url.endswith("InstallESD.dmg"):
base_system = {"url": url, "size": size}
return install_assistant, base_system, total_size
def list_macos_installers():
"""Gibt (items, catalog_url) zurueck. items ist eine sortierte Liste von dicts."""
url = find_latest_catalog_url()
products = _fetch_catalog_products(url)
results = []
for pid, product in products.items():
if not _is_macos_installer(product):
continue
ia, base, size = _find_packages(product)
if not ia and not base:
continue
title, version = _fetch_product_metadata(product)
post_date = product.get("PostDate")
results.append(
{
"id": pid,
"title": title or "macOS (Name unbekannt)",
"version": version or "0",
"post_date": post_date.isoformat() if post_date else None,
"size_bytes": size,
# "recovery" = Big Sur+ (InstallAssistant.pkg, Extraktion noetig, experimentell)
# "legacy_dmg" = bis Catalina (direktes dd-Image, robust)
"mode": "recovery" if ia else "legacy_dmg",
"install_assistant_url": ia["url"] if ia else None,
"base_system_url": base["url"] if base else None,
}
)
def _sort_key(item):
try:
return tuple(int(x) for x in item["version"].split("."))
except Exception: # noqa: BLE001
return (0,)
results.sort(key=_sort_key, reverse=True)
return results, url
backend/devices.py
+53
View File
@@ -0,0 +1,53 @@
"""USB-Block-Devices des Hosts auflisten (via lsblk, /dev + /run/udev gemountet)."""
import json
import subprocess
def list_usb_sticks():
out = subprocess.run(
["lsblk", "-J", "-b", "-o", "NAME,SIZE,MODEL,VENDOR,TRAN,RM,TYPE,MOUNTPOINT"],
capture_output=True,
text=True,
check=True,
).stdout
data = json.loads(out)
root = get_root_disk_name()
sticks = []
for dev in data.get("blockdevices", []):
if dev.get("type") != "disk":
continue
is_removable = str(dev.get("rm")) in ("1", "True", "true")
is_usb_tran = dev.get("tran") == "usb"
if not (is_removable or is_usb_tran):
continue
name = dev["name"]
mounted = bool(dev.get("mountpoint")) or any(
c.get("mountpoint") for c in (dev.get("children") or [])
)
sticks.append(
{
"name": name,
"path": f"/dev/{name}",
"size_bytes": int(dev.get("size") or 0),
"model": (dev.get("model") or "").strip(),
"vendor": (dev.get("vendor") or "").strip(),
"transport": dev.get("tran"),
"mounted": mounted,
"is_root_disk": name == root,
}
)
return sticks
def get_root_disk_name():
"""Name (ohne /dev/) der Platte, auf der / liegt -- fuer Schutz vor Selbstzerstoerung."""
try:
src = subprocess.run(
["findmnt", "-n", "-o", "SOURCE", "/"], capture_output=True, text=True, check=True
).stdout.strip()
pk = subprocess.run(
["lsblk", "-no", "PKNAME", src], capture_output=True, text=True
).stdout.strip()
return pk or src.replace("/dev/", "")
except Exception: # noqa: BLE001
return None
backend/installer.py
+266
View File
@@ -0,0 +1,266 @@
"""
Job-Engine: Download eines macOS-Installers + Beschreiben eines USB-Sticks.
Zwei Pfade, unterschiedlich robust:
legacy_dmg (macOS bis Catalina/10.15): Apple liefert ein fertiges,
bootfaehiges BaseSystem.dmg/InstallESD.dmg. Wir wandeln es mit
dmg2img in ein rohes Image und schreiben es 1:1 mit dd auf den
Stick. Etablierter, zuverlaessiger Weg.
recovery (macOS ab Big Sur/11): Apple liefert nur noch ein
InstallAssistant.pkg. Das eigentliche createinstallmedia laeuft
nur unter echtem macOS. Wir extrahieren das .pkg (xar -> pbzx
-> cpio) und suchen darin ein SharedSupport.dmg/BaseSystem.dmg,
das wir dann genauso per dd schreiben. EXPERIMENTELL: Apple hat
den internen Aufbau von InstallAssistant.pkg mehrfach leicht
geaendert, das kann brechen. Klar als "experimentell" markiert.
Sicherheits-Leitplanken:
- Ziel-Device wird gegen die Root-Disk des Hosts geprueft (Refuse).
- Ziel-Device muss "removable" oder USB-Transport sein (Refuse sonst).
- Aufrufer muss den Device-Pfad im Request nochmal exakt bestaetigen.
"""
import logging
import os
import shutil
import subprocess
import tempfile
import threading
import time
import uuid
from queue import Queue, Empty
import requests
from devices import list_usb_sticks
log = logging.getLogger("installer")
DOWNLOAD_DIR = "/data/downloads"
JOBS: dict[str, "Job"] = {}
class JobAborted(Exception):
pass
class Job:
def __init__(self, product: dict, device_path: str):
self.id = str(uuid.uuid4())
self.product = product
self.device_path = device_path
self.status = "queued" # queued -> running -> done | failed | aborted
self.phase = "warten"
self.progress = 0 # 0-100 im aktuellen Phase
self.error = None
self._q: Queue = Queue()
self._lock = threading.Lock()
self.history = []
def emit(self, phase=None, progress=None, msg=None, status=None):
with self._lock:
if phase is not None:
self.phase = phase
if progress is not None:
self.progress = progress
if status is not None:
self.status = status
event = {
"phase": self.phase,
"progress": self.progress,
"status": self.status,
"msg": msg,
"ts": time.time(),
}
self.history.append(event)
self._q.put(event)
def stream(self):
# erst die bisherige Historie nachliefern, dann live weiter
for event in list(self.history):
yield event
while True:
try:
yield self._q.get(timeout=30)
except Empty:
if self.status in ("done", "failed", "aborted"):
return
yield {"phase": self.phase, "progress": self.progress, "status": self.status, "msg": None, "ts": time.time()}
def snapshot(self):
with self._lock:
return {
"id": self.id,
"status": self.status,
"phase": self.phase,
"progress": self.progress,
"error": self.error,
}
def create_job(product: dict, device_path: str, confirm_path: str) -> Job:
if confirm_path != device_path:
raise ValueError("Bestaetigungs-Pfad stimmt nicht mit dem Ziel-Geraet ueberein.")
sticks = list_usb_sticks()
match = next((s for s in sticks if s["path"] == device_path), None)
if not match:
raise ValueError(f"Geraet {device_path} nicht in der aktuellen USB-Liste gefunden.")
if match["is_root_disk"]:
raise ValueError("Verweigert: das ist die System-Platte des Hosts.")
if match["mounted"]:
raise ValueError("Geraet ist aktuell gemountet -- bitte vorher aushaengen.")
if not (match["transport"] == "usb"):
raise ValueError("Geraet ist nicht als USB-Transport erkannt -- Sicherheitsstop.")
job = Job(product, device_path)
JOBS[job.id] = job
t = threading.Thread(target=_run_job, args=(job,), daemon=True)
t.start()
return job
def _run(job: Job, cmd: list[str], phase: str, progress_from_stderr=False):
log.info("job %s: %s", job.id, " ".join(cmd))
proc = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, bufsize=1)
for line in proc.stdout:
line = line.strip()
if line:
job.emit(phase=phase, msg=line)
ret = proc.wait()
if ret != 0:
raise RuntimeError(f"Befehl fehlgeschlagen ({ret}): {' '.join(cmd)}")
def _download(job: Job, url: str, dest_path: str):
job.emit(phase="download", progress=0, msg=f"Lade {url}")
with requests.get(url, stream=True, timeout=30) as r:
r.raise_for_status()
total = int(r.headers.get("Content-Length", 0)) or None
done = 0
last_pct = -1
with open(dest_path, "wb") as f:
for chunk in r.iter_content(chunk_size=1024 * 1024 * 4):
if not chunk:
continue
f.write(chunk)
done += len(chunk)
if total:
pct = int(done * 100 / total)
if pct != last_pct:
job.emit(phase="download", progress=pct)
last_pct = pct
job.emit(phase="download", progress=100, msg="Download fertig")
def _dmg_to_raw(job: Job, dmg_path: str) -> str:
raw_path = dmg_path + ".img"
job.emit(phase="konvertieren", progress=0, msg="dmg2img: DMG -> rohes Image")
_run(job, ["dmg2img", "-i", dmg_path, "-o", raw_path], phase="konvertieren")
return raw_path
def _write_raw_to_device(job: Job, raw_path: str, device_path: str):
job.emit(phase="schreiben", progress=0, msg=f"Schreibe {raw_path} -> {device_path} (dd)")
_run(job, ["wipefs", "-a", device_path], phase="schreiben")
size = os.path.getsize(raw_path)
cmd = ["dd", f"if={raw_path}", f"of={device_path}", "bs=4M", "status=progress", "conv=fsync"]
proc = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, bufsize=1)
for line in proc.stdout:
line = line.strip()
if not line:
continue
job.emit(phase="schreiben", msg=line)
# dd status=progress schreibt Zeilen wie "1234567890 bytes (1.2 GB, 1.1 GiB) copied, ..."
try:
bytes_done = int(line.split(" ")[0])
pct = min(100, int(bytes_done * 100 / size))
job.emit(phase="schreiben", progress=pct)
except (ValueError, IndexError, ZeroDivisionError):
pass
ret = proc.wait()
if ret != 0:
raise RuntimeError("dd fehlgeschlagen")
job.emit(phase="schreiben", progress=100, msg="dd fertig, sync...")
subprocess.run(["sync"])
def _extract_recovery_dmg(job: Job, pkg_path: str, work_dir: str) -> str:
"""Experimentell: InstallAssistant.pkg -> xar -> pbzx -> cpio, sucht SharedSupport/BaseSystem.dmg."""
job.emit(phase="extrahieren", progress=0, msg="xar: InstallAssistant.pkg entpacken (experimentell)")
extract_dir = os.path.join(work_dir, "xar")
os.makedirs(extract_dir, exist_ok=True)
_run(job, ["xar", "-xf", pkg_path, "-C", extract_dir], phase="extrahieren")
payloads = []
for root, _dirs, files in os.walk(extract_dir):
for fn in files:
if fn == "Payload":
payloads.append(os.path.join(root, fn))
if not payloads:
raise RuntimeError("Kein Payload in InstallAssistant.pkg gefunden -- Apple-Struktur hat sich vermutlich geaendert.")
cpio_dir = os.path.join(work_dir, "cpio")
os.makedirs(cpio_dir, exist_ok=True)
for payload in payloads:
job.emit(phase="extrahieren", msg=f"pbzx+cpio: {payload}")
# pbzx dekomprimiert das payload, cpio packt den entstandenen Stream aus
pbzx_proc = subprocess.Popen(["pbzx", "-n", payload], stdout=subprocess.PIPE)
cpio_proc = subprocess.Popen(
["cpio", "-idm"], cwd=cpio_dir, stdin=pbzx_proc.stdout,
stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True,
)
pbzx_proc.stdout.close()
for line in cpio_proc.stdout:
line = line.strip()
if line:
job.emit(phase="extrahieren", msg=line)
cpio_proc.wait()
pbzx_proc.wait()
dmg_candidates = []
for root, _dirs, files in os.walk(cpio_dir):
for fn in files:
if fn in ("SharedSupport.dmg", "BaseSystem.dmg"):
dmg_candidates.append(os.path.join(root, fn))
if not dmg_candidates:
raise RuntimeError(
"Kein SharedSupport.dmg/BaseSystem.dmg im entpackten Installer gefunden. "
"Das ist der bekannte fragile Punkt bei macOS Big Sur+ ohne echten Mac."
)
# SharedSupport.dmg bevorzugen (enthaelt das vollstaendige Recovery-System)
dmg_candidates.sort(key=lambda p: 0 if "SharedSupport" in p else 1)
return dmg_candidates[0]
def _run_job(job: Job):
os.makedirs(DOWNLOAD_DIR, exist_ok=True)
work_dir = tempfile.mkdtemp(dir=DOWNLOAD_DIR)
try:
job.emit(status="running", phase="start", progress=0, msg="Job gestartet")
product = job.product
mode = product["mode"]
if mode == "legacy_dmg":
url = product["base_system_url"]
dmg_path = os.path.join(work_dir, "BaseSystem.dmg")
_download(job, url, dmg_path)
raw_path = _dmg_to_raw(job, dmg_path)
_write_raw_to_device(job, raw_path, job.device_path)
else:
job.emit(phase="hinweis", msg="Big Sur+ Pfad ist experimentell (siehe README).")
url = product["install_assistant_url"]
pkg_path = os.path.join(work_dir, "InstallAssistant.pkg")
_download(job, url, pkg_path)
dmg_path = _extract_recovery_dmg(job, pkg_path, work_dir)
raw_path = _dmg_to_raw(job, dmg_path)
_write_raw_to_device(job, raw_path, job.device_path)
job.emit(status="done", phase="fertig", progress=100, msg="Stick ist fertig.")
except Exception as exc: # noqa: BLE001
log.exception("job %s failed", job.id)
job.error = str(exc)
job.emit(status="failed", msg=f"FEHLER: {exc}")
finally:
shutil.rmtree(work_dir, ignore_errors=True)
backend/requirements.txt
+2
View File
@@ -0,0 +1,2 @@
Flask==3.0.3
requests==2.32.3