353 lines
17 KiB
Python
353 lines
17 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
@file build_bank.py
|
||
@brief Prüft alle Fragendateien unter data/, erzeugt die verschlüsselte js/bank.js und optional ein Export-Verzeichnis
|
||
@author Björn Runden <br@runden.de>
|
||
@owner Björn Runden <br@runden.de>
|
||
@assist KI-unterstützt (Claude Opus 5.5), geprüft von Björn Runden <br@runden.de>
|
||
@version 0.3.0 (29.10.2026)
|
||
@usage build_bank.py [-d] [-q] [-n] [-c] [--approved-only] [-o js/bank.js] [-e [DIR]] [-b [FILE]]
|
||
|
||
Aufbau von data/: je Prüfung ein Ordner mit exam.json und beliebig vielen questions_*.json (Quelltext, Klartext).
|
||
js/bank.js enthält die Fragen nur ChaCha20-verschlüsselt (RFC 8439). Der Schlüssel wird aus einem PSK abgeleitet
|
||
(Umgebung LPIC_SIM_PSK, sonst tools/.psk, sonst zufällig erzeugt) und steht als „build“-Kennung in bank.js –
|
||
das verhindert Klartext auf der Platte, ist aber kein Schutz gegen gezielte Analyse (der Browser braucht den Schlüssel).
|
||
Mit -n/--dry-run bzw. -c/--check wird nur geprüft und die Abdeckung je Objective ausgegeben.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import base64
|
||
import hashlib
|
||
import json
|
||
import logging
|
||
import os
|
||
import re
|
||
import secrets
|
||
import shutil
|
||
import struct
|
||
import sys
|
||
from datetime import datetime, timezone
|
||
from pathlib import Path
|
||
|
||
BASE = Path(__file__).resolve().parent.parent
|
||
DATA = BASE / "data"
|
||
OUT_DEFAULT = BASE / "js" / "bank.js"
|
||
PSK_FILE = BASE / "tools" / ".psk"
|
||
EXPORT_DEFAULT = BASE / "dist"
|
||
EXPORT_MARKER = ".lpic-sim-export"
|
||
EXPORT_ITEMS = ["index.html", "css", "js", "vendor"]
|
||
|
||
OPTION_IDS = ["a", "b", "c", "d", "e"]
|
||
QID_RE = re.compile(r"^(\d{3}\.\d+)-\d{3}$")
|
||
QUESTION_KEYS = {"id", "objective", "type", "lang", "difficulty", "question", "code", "options",
|
||
"answers", "case_sensitive", "explanation", "references", "review"}
|
||
REVIEW_STATES = {"pending", "approved", "rejected"}
|
||
|
||
log = logging.getLogger("build_bank")
|
||
|
||
|
||
# ------------------------------------------------------------------ ChaCha20 (RFC 8439), nur Standardbibliothek
|
||
|
||
def _chacha20_block(key_words: tuple, counter: int, nonce_words: tuple) -> bytes:
|
||
"""Ein 64-Byte-Keystream-Block nach RFC 8439, Abschnitt 2.3."""
|
||
state = [0x61707865, 0x3320646E, 0x79622D32, 0x6B206574, *key_words, counter & 0xFFFFFFFF, *nonce_words]
|
||
x = state[:]
|
||
|
||
def qr(a: int, b: int, c: int, d: int) -> None:
|
||
x[a] = (x[a] + x[b]) & 0xFFFFFFFF; x[d] ^= x[a]; x[d] = ((x[d] << 16) | (x[d] >> 16)) & 0xFFFFFFFF
|
||
x[c] = (x[c] + x[d]) & 0xFFFFFFFF; x[b] ^= x[c]; x[b] = ((x[b] << 12) | (x[b] >> 20)) & 0xFFFFFFFF
|
||
x[a] = (x[a] + x[b]) & 0xFFFFFFFF; x[d] ^= x[a]; x[d] = ((x[d] << 8) | (x[d] >> 24)) & 0xFFFFFFFF
|
||
x[c] = (x[c] + x[d]) & 0xFFFFFFFF; x[b] ^= x[c]; x[b] = ((x[b] << 7) | (x[b] >> 25)) & 0xFFFFFFFF
|
||
|
||
for _ in range(10):
|
||
qr(0, 4, 8, 12); qr(1, 5, 9, 13); qr(2, 6, 10, 14); qr(3, 7, 11, 15)
|
||
qr(0, 5, 10, 15); qr(1, 6, 11, 12); qr(2, 7, 8, 13); qr(3, 4, 9, 14)
|
||
return struct.pack("<16L", *((x[i] + state[i]) & 0xFFFFFFFF for i in range(16)))
|
||
|
||
|
||
def chacha20(key: bytes, nonce: bytes, data: bytes, counter: int = 1) -> bytes:
|
||
"""Ver-/Entschlüsselt data mit ChaCha20 (32-Byte-Schlüssel, 12-Byte-Nonce). Returns: Ergebnis gleicher Länge."""
|
||
kw, nw = struct.unpack("<8L", key), struct.unpack("<3L", nonce)
|
||
out = bytearray(len(data))
|
||
for off in range(0, len(data), 64):
|
||
block = _chacha20_block(kw, counter + off // 64, nw)
|
||
chunk = data[off:off + 64]
|
||
out[off:off + len(chunk)] = bytes(a ^ b for a, b in zip(chunk, block))
|
||
return bytes(out)
|
||
|
||
|
||
def load_psk() -> str:
|
||
"""PSK aus LPIC_SIM_PSK oder tools/.psk; fehlt beides, wird einer erzeugt und in tools/.psk abgelegt."""
|
||
env = os.environ.get("LPIC_SIM_PSK", "").strip()
|
||
if env:
|
||
return env
|
||
if PSK_FILE.is_file():
|
||
psk = PSK_FILE.read_text(encoding="utf-8").strip()
|
||
if psk:
|
||
return psk
|
||
psk = secrets.token_urlsafe(24)
|
||
PSK_FILE.write_text(psk + "\n", encoding="utf-8")
|
||
PSK_FILE.chmod(0o600)
|
||
log.info("neuer PSK erzeugt: %s (nicht weitergeben)", PSK_FILE.relative_to(BASE))
|
||
return psk
|
||
|
||
|
||
def derive_key(psk: str) -> bytes:
|
||
"""Schlüssel = SHA-256 über den PSK mit festem Kontext; so taucht der PSK selbst nirgends auf."""
|
||
return hashlib.sha256(b"lpic-sim/bank/v1:" + psk.encode("utf-8")).digest()
|
||
|
||
|
||
# ------------------------------------------------------------------ Prüfung der Fragen
|
||
|
||
def load_json(path: Path) -> dict:
|
||
"""Liest eine JSON-Datei; Fehler werden mit Dateiname gemeldet."""
|
||
try:
|
||
return json.loads(path.read_text(encoding="utf-8"))
|
||
except json.JSONDecodeError as exc:
|
||
raise ValueError(f"{path.relative_to(BASE)}: ungültiges JSON ({exc})") from exc
|
||
|
||
|
||
def review_state(q: dict) -> str:
|
||
"""Prüfstatus einer Frage; ohne Eintrag 'pending'."""
|
||
return (q.get("review") or {}).get("status", "pending")
|
||
|
||
|
||
def check_question(q: dict, objectives: set[str], where: str) -> list[str]:
|
||
"""Prüft eine Frage gegen die Erstellungsregeln. Returns: Liste der Fehlermeldungen."""
|
||
err: list[str] = []
|
||
qid = q.get("id", "?")
|
||
p = f"{where} [{qid}]"
|
||
unknown = set(q) - QUESTION_KEYS
|
||
if unknown:
|
||
err.append(f"{p}: unbekannte Schlüssel {sorted(unknown)}")
|
||
m = QID_RE.match(qid)
|
||
if not m:
|
||
err.append(f"{p}: id muss <objective>-<NNN> sein")
|
||
elif m.group(1) != q.get("objective"):
|
||
err.append(f"{p}: id passt nicht zu objective {q.get('objective')}")
|
||
if q.get("objective") not in objectives:
|
||
err.append(f"{p}: objective {q.get('objective')} gibt es in der Prüfung nicht")
|
||
for key in ("question", "explanation"):
|
||
if not str(q.get(key, "")).strip():
|
||
err.append(f"{p}: '{key}' fehlt oder ist leer")
|
||
if review_state(q) not in REVIEW_STATES:
|
||
err.append(f"{p}: review.status muss pending, approved oder rejected sein")
|
||
qtype = q.get("type")
|
||
if qtype in ("single", "multi"):
|
||
opts = q.get("options") or []
|
||
if [o.get("id") for o in opts] != OPTION_IDS:
|
||
err.append(f"{p}: genau 5 Optionen mit ids a–e in dieser Reihenfolge erforderlich")
|
||
texts = [o.get("text", "").strip() for o in opts]
|
||
if len(set(texts)) != len(texts):
|
||
err.append(f"{p}: doppelte Optionstexte")
|
||
for o in opts:
|
||
if not o.get("text", "").strip() or not o.get("explanation", "").strip():
|
||
err.append(f"{p}: Option {o.get('id')} ohne Text oder Erläuterung")
|
||
if not isinstance(o.get("correct"), bool):
|
||
err.append(f"{p}: Option {o.get('id')}: 'correct' muss true/false sein")
|
||
n_ok = sum(1 for o in opts if o.get("correct") is True)
|
||
if qtype == "single" and n_ok != 1:
|
||
err.append(f"{p}: single braucht genau 1 richtige Option (hat {n_ok})")
|
||
if qtype == "multi" and not 2 <= n_ok <= 4:
|
||
err.append(f"{p}: multi braucht 2–4 richtige Optionen (hat {n_ok})")
|
||
if "answers" in q:
|
||
err.append(f"{p}: 'answers' nur bei type fill")
|
||
elif qtype == "fill":
|
||
if not q.get("answers") or not all(str(a).strip() for a in q["answers"]):
|
||
err.append(f"{p}: fill braucht nicht leere 'answers'")
|
||
if "options" in q:
|
||
err.append(f"{p}: 'options' nicht bei type fill")
|
||
else:
|
||
err.append(f"{p}: type muss single, multi oder fill sein")
|
||
return err
|
||
|
||
|
||
def load_exam(exam_dir: Path, approved_only: bool) -> tuple[dict, list[str]]:
|
||
"""Lädt exam.json und alle questions_*.json eines Prüfungsordners. Returns: (Prüfung, Fehler)."""
|
||
exam_doc = load_json(exam_dir / "exam.json")
|
||
exam = dict(exam_doc["exam"])
|
||
exam["topics"] = exam_doc["topics"]
|
||
objectives = {o["id"] for t in exam["topics"] for o in t["objectives"]}
|
||
errors: list[str] = []
|
||
questions: list[dict] = []
|
||
seen: set[str] = set()
|
||
states = {s: 0 for s in sorted(REVIEW_STATES)}
|
||
for qfile in sorted(exam_dir.glob("questions_*.json")):
|
||
doc = load_json(qfile)
|
||
where = str(qfile.relative_to(BASE))
|
||
if doc.get("format") != "lpic-sim-questions":
|
||
errors.append(f"{where}: format muss 'lpic-sim-questions' sein")
|
||
if doc.get("exam") != exam["id"]:
|
||
errors.append(f"{where}: exam '{doc.get('exam')}' passt nicht zu '{exam['id']}'")
|
||
for q in doc.get("questions", []):
|
||
errors += check_question(q, objectives, where)
|
||
if q.get("id") in seen:
|
||
errors.append(f"{where}: doppelte id {q.get('id')}")
|
||
seen.add(q.get("id"))
|
||
state = review_state(q)
|
||
states[state] = states.get(state, 0) + 1
|
||
if state == "rejected" or (approved_only and state != "approved"):
|
||
continue
|
||
q = {k: v for k, v in q.items() if k != "review"} # Prüfvermerke gehören nicht in die Bank
|
||
q.setdefault("lang", exam.get("lang", "de"))
|
||
questions.append(q)
|
||
questions.sort(key=lambda q: (tuple(int(x) for x in q["objective"].split(".")), q["id"]))
|
||
exam["questions_bank"] = questions
|
||
exam["review_states"] = states
|
||
return exam, errors
|
||
|
||
|
||
def coverage(exam: dict) -> None:
|
||
"""Gibt Prüfstatus und die Anzahl der Fragen je Objective im Verhältnis zur Gewichtung aus."""
|
||
counts: dict[str, int] = {}
|
||
for q in exam["questions_bank"]:
|
||
counts[q["objective"]] = counts.get(q["objective"], 0) + 1
|
||
st = exam["review_states"]
|
||
log.info("%s: %d Fragen in der Bank (Prüfstatus: %d approved, %d pending, %d rejected)", exam["id"],
|
||
len(exam["questions_bank"]), st.get("approved", 0), st.get("pending", 0), st.get("rejected", 0))
|
||
for t in exam["topics"]:
|
||
for o in t["objectives"]:
|
||
n = counts.get(o["id"], 0)
|
||
flag = "" if n >= o["weight"] else " <- weniger Fragen als Gewicht (Prüfungsmodus füllt auf)"
|
||
log.info(" %-6s Gewicht %d Fragen %3d%s", o["id"], o["weight"], n, flag)
|
||
|
||
|
||
# ------------------------------------------------------------------ Ausgabe
|
||
|
||
def write_bank(exams: list[dict], output: Path) -> None:
|
||
"""Schreibt bank.js: Metadaten im Klartext, Prüfungen und Fragen ChaCha20-verschlüsselt (Base64)."""
|
||
generated = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
|
||
# exam["questions"] bleibt die Fragenzahl der Prüfung, die Fragen selbst stehen unter "bank"
|
||
plain = json.dumps({"exams": [{**{k: v for k, v in e.items() if k not in ("questions_bank", "review_states")},
|
||
"bank": e["questions_bank"]} for e in exams]},
|
||
ensure_ascii=False, separators=(",", ":")).encode("utf-8")
|
||
key, nonce = derive_key(load_psk()), secrets.token_bytes(12)
|
||
bank = {
|
||
"generated": generated,
|
||
"build": key.hex(), # Schlüssel, getarnt als Build-Kennung
|
||
"rev": nonce.hex(), # Nonce
|
||
"payload": base64.b64encode(chacha20(key, nonce, plain)).decode("ascii"),
|
||
}
|
||
header = ("/**\n * @file bank.js\n * @brief Fragenbank (verschlüsselt) – ERZEUGT von tools/build_bank.py, nicht von Hand bearbeiten\n"
|
||
" * @author Björn Runden <br@runden.de>\n * @owner Björn Runden <br@runden.de>\n"
|
||
f" * @version {generated}\n */\n")
|
||
output.write_text(header + "window.LPIC_BANK = " + json.dumps(bank, indent=1) + ";\n", encoding="utf-8")
|
||
|
||
|
||
def export_site(target: Path, dry_run: bool) -> None:
|
||
"""Kopiert nur die Laufzeitdateien (ohne data/, tools/, tests/) nach target.
|
||
Ein vorhandenes Ziel wird nur ersetzt, wenn es von hier stammt (Markierungsdatei)."""
|
||
target = target.resolve()
|
||
if target == BASE or BASE in target.parents and target.name in EXPORT_ITEMS + ["data", "tools", "tests"]:
|
||
raise ValueError(f"Export-Ziel {target} liegt im Quellbaum")
|
||
if target.exists() and any(target.iterdir()) and not (target / EXPORT_MARKER).exists():
|
||
raise ValueError(f"Export-Ziel {target} ist nicht leer und kein früherer Export – bitte prüfen")
|
||
if dry_run:
|
||
log.info("Trockenlauf: würde %s nach %s exportieren", ", ".join(EXPORT_ITEMS), target)
|
||
return
|
||
if target.exists():
|
||
shutil.rmtree(target)
|
||
target.mkdir(parents=True)
|
||
for item in EXPORT_ITEMS:
|
||
src = BASE / item
|
||
(shutil.copytree if src.is_dir() else shutil.copy2)(src, target / item)
|
||
(target / EXPORT_MARKER).write_text("Export von simulator/tools/build_bank.py – Verzeichnis wird beim nächsten Export ersetzt\n",
|
||
encoding="utf-8")
|
||
log.info("Export nach %s geschrieben (Start: python3 -m http.server 8080 -d %s)", target, target)
|
||
|
||
|
||
def bundle_excluded(rel: Path) -> bool:
|
||
"""Nicht ins Prüferpaket: PSK, Erzeugtes (Export, Pakete), Caches, Sicherungen."""
|
||
parts = rel.parts
|
||
return (rel.as_posix() == "tools/.psk" or parts[0] in ("dist", "release")
|
||
or "__pycache__" in parts or rel.name.endswith(".orig"))
|
||
|
||
|
||
def write_bundle(target: Path, dry_run: bool) -> None:
|
||
"""Packt den kompletten Quellstand (Fragen, Doku, Compiler, Tests, Laufzeit) als tar.gz mit Wurzel lpic-sim/."""
|
||
import tarfile
|
||
files = sorted(p for p in BASE.rglob("*") if p.is_file() and not bundle_excluded(p.relative_to(BASE)))
|
||
if dry_run:
|
||
log.info("Trockenlauf: würde %d Dateien nach %s packen", len(files), target)
|
||
return
|
||
target.parent.mkdir(parents=True, exist_ok=True)
|
||
with tarfile.open(target, "w:gz") as tar:
|
||
for p in files:
|
||
tar.add(p, arcname=f"lpic-sim/{p.relative_to(BASE).as_posix()}", recursive=False)
|
||
log.info("Prüferpaket %s geschrieben (%d Dateien, %d KiB)", target, len(files), target.stat().st_size // 1024)
|
||
|
||
|
||
def main() -> int:
|
||
ap = argparse.ArgumentParser(description="Fragenbank prüfen, verschlüsselte js/bank.js erzeugen, optional exportieren")
|
||
ap.add_argument("-d", "--debug", action="store_true", help="Debug-Ausgaben")
|
||
ap.add_argument("-q", "--quiet", action="store_true", help="nur Fehler ausgeben")
|
||
ap.add_argument("-n", "--dry-run", action="store_true", help="nur prüfen, nichts schreiben")
|
||
ap.add_argument("-c", "--check", action="store_true", help="wie --dry-run, mit Abdeckungsbericht")
|
||
ap.add_argument("--approved-only", action="store_true", help="nur Fragen mit review.status approved übernehmen")
|
||
ap.add_argument("-o", "--output", type=Path, default=OUT_DEFAULT, help="Zieldatei (Default js/bank.js)")
|
||
ap.add_argument("-e", "--export", type=Path, nargs="?", const=EXPORT_DEFAULT, metavar="DIR",
|
||
help=f"Laufzeitdateien nach DIR exportieren (Default {EXPORT_DEFAULT.relative_to(BASE)})")
|
||
ap.add_argument("-b", "--bundle", type=Path, nargs="?", metavar="FILE",
|
||
const=BASE / "release" / f"lpic-sim_{datetime.now():%Y-%m-%d}.tar.gz",
|
||
help="Prüferpaket (tar.gz, ohne PSK und Erzeugtes) schreiben; Default release/lpic-sim_<datum>.tar.gz")
|
||
args = ap.parse_args()
|
||
level = logging.DEBUG if args.debug else logging.ERROR if args.quiet else logging.INFO
|
||
logging.basicConfig(level=level, format="%(levelname)s %(message)s")
|
||
|
||
exams: list[dict] = []
|
||
errors: list[str] = []
|
||
try:
|
||
for exam_file in sorted(DATA.glob("*/exam.json")):
|
||
exam, errs = load_exam(exam_file.parent, args.approved_only)
|
||
exams.append(exam)
|
||
errors += errs
|
||
except (ValueError, KeyError) as exc:
|
||
log.error("%s", exc)
|
||
return 1
|
||
for e in errors:
|
||
log.error("%s", e)
|
||
if errors:
|
||
log.error("%d Fehler – bank.js nicht geschrieben", len(errors))
|
||
return 1
|
||
exams.sort(key=lambda e: (e.get("level", ""), e.get("order", 0)))
|
||
total = sum(len(e["questions_bank"]) for e in exams)
|
||
if args.check or args.debug:
|
||
for exam in exams:
|
||
coverage(exam)
|
||
if args.dry_run or args.check:
|
||
log.info("Prüfung ok: %d Prüfungen, %d Fragen", len(exams), total)
|
||
if args.export:
|
||
try:
|
||
export_site(args.export, True)
|
||
except ValueError as exc:
|
||
log.error("%s", exc)
|
||
return 1
|
||
if args.bundle:
|
||
write_bundle(args.bundle, True)
|
||
return 0
|
||
if total == 0:
|
||
log.error("keine Fragen für die Bank (bei --approved-only: noch nichts freigegeben?)")
|
||
return 1
|
||
|
||
write_bank(exams, args.output)
|
||
log.info("%s geschrieben (verschlüsselt): %d Prüfungen, %d Fragen", args.output, len(exams), total)
|
||
if args.export:
|
||
try:
|
||
export_site(args.export, False)
|
||
except (ValueError, OSError) as exc:
|
||
log.error("Export fehlgeschlagen: %s", exc)
|
||
return 1
|
||
if args.bundle:
|
||
try:
|
||
write_bundle(args.bundle, False)
|
||
except OSError as exc:
|
||
log.error("Prüferpaket fehlgeschlagen: %s", exc)
|
||
return 1
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|