#!/usr/bin/env python3 """FKRPG audit: PACKWIZ is source of truth. Inspects Packwiz project, classifies mods via mod-version-policy.json, checks index integrity, hashes, drift (LOCAL vs PACKWIZ vs PRODUCTION), detects outdated optimization/worldgen mods, and generates reports. Never auto-updates FROZEN mods — reports BLOCKED if optimization requires it. Stdlib-only, Fish-friendly text + JSON output. """ import argparse import glob import hashlib import json import os import re import sys import urllib.request import urllib.error from collections import Counter, defaultdict REPO = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) SERVER_PACK = os.path.join(REPO, "server_pack") CLIENT_PACK = os.path.join(REPO, "client_pack") POLICY_PATH = os.path.join(REPO, "mod-version-policy.json") MANIFEST_PATH = os.path.join(REPO, "optimization-manifest.json") LOCAL_SERVER = os.path.expanduser("~/.auto-mcs/Servers/FKRPG") # Also check docker data volume if present PRODUCTION_HINT = os.path.join(REPO, "data") PW_TOML = re.compile(r"^(\w[\w.-]*)\s*=\s*(?:\"([^\"]*)\"|([0-9]+))", re.M) SECTION = re.compile(r"^\[\[?([\w.\-]+)\]\]?$", re.M) API = "https://api.modrinth.com/v2" def parse_toml(path): data = {} cur = data with open(path, encoding="utf-8") as fh: for line in fh: line = line.strip() if not line or line.startswith("#"): continue m = SECTION.match(line) if m: name = m.group(1) if line.startswith("[["): data.setdefault(name, []) cur = {} data[name].append(cur) else: key = name.replace("-", "_").replace(".", "_") data.setdefault(key, {}) cur = data[key] continue m = PW_TOML.match(line) if m: key = m.group(1).replace("-", "_") val = m.group(2) if m.group(2) is not None else m.group(3) cur[key] = val return data def load_policy(): if not os.path.exists(POLICY_PATH): return {} with open(POLICY_PATH) as f: return json.load(f) def load_manifest(): if not os.path.exists(MANIFEST_PATH): return {} with open(MANIFEST_PATH) as f: return json.load(f) def mod_id_from_pw(path): d = parse_toml(path) return d.get("update", {}).get("modrinth", {}).get("mod_id") or d.get("update_modrinth", {}).get("mod_id") def scan_pack(pack_dir): mods = {} for f in sorted(glob.glob(os.path.join(pack_dir, "mods", "*.pw.toml"))): d = parse_toml(f) slug = os.path.basename(f).replace(".pw.toml", "") mods[slug] = { "slug": slug, "path": f, "filename": d.get("filename", ""), "side": d.get("side", "both"), "download_url": d.get("download", {}).get("url", ""), "hash": d.get("download", {}).get("hash", ""), "hash_format": d.get("download", {}).get("hash_format", "sha512"), "mod_id": d.get("update", {}).get("modrinth", {}).get("mod_id") or d.get("update_modrinth", {}).get("mod_id", ""), "version_id": d.get("update", {}).get("modrinth", {}).get("version") or d.get("update_modrinth", {}).get("version", ""), "raw": d, } pack_toml = parse_toml(os.path.join(pack_dir, "pack.toml")) if os.path.exists(os.path.join(pack_dir, "pack.toml")) else {} index = parse_toml(os.path.join(pack_dir, "index.toml")) if os.path.exists(os.path.join(pack_dir, "index.toml")) else {} return mods, pack_toml, index def check_index(pack_dir, mods, index): indexed = {e.get("file") for e in index.get("files", []) if e.get("file")} pw_files = {f"mods/{os.path.basename(f)}" for f in glob.glob(os.path.join(pack_dir, "mods", "*.pw.toml"))} missing = sorted(pw_files - indexed) extra = sorted(indexed - pw_files - {f for f in indexed if not f.startswith("mods/")}) # only care about mods extra_mods = [x for x in extra if x.startswith("mods/")] return missing, extra_mods def sha256_file(path): h = hashlib.sha256() with open(path, "rb") as f: for chunk in iter(lambda: f.read(8192), b""): h.update(chunk) return h.hexdigest() def check_hashes(mods): errors = [] for slug, m in mods.items(): hf = m["hash_format"] hv = m["hash"] if not hv: errors.append(f"{slug}: missing hash") elif not re.fullmatch(r"[0-9a-f]+", hv): errors.append(f"{slug}: hash not hex ({hf})") if m["download_url"].startswith("http://"): errors.append(f"{slug}: insecure http url") return errors def get_local_mods(): if not os.path.isdir(LOCAL_SERVER): return None, "not found" mods_dir = os.path.join(LOCAL_SERVER, "mods") if not os.path.isdir(mods_dir): return {}, "no mods dir" jars = [os.path.basename(f) for f in glob.glob(os.path.join(mods_dir, "*.jar"))] return set(jars), None def get_production_hint(): # Check docker compose file-mount hints, or data/mods candidates = [ os.path.join(REPO, "data", "mods"), os.path.join(REPO, "server-pack.mrpack"), ] info = {} for c in candidates: if os.path.exists(c): if os.path.isdir(c): jars = [os.path.basename(f) for f in glob.glob(os.path.join(c, "*.jar"))] info[c] = {"type": "dir", "count": len(jars), "sample": jars[:5]} else: info[c] = {"type": "file", "sha256": sha256_file(c)[:16] + "..." if os.path.isfile(c) else "?"} return info def fetch_modrinth_latest(mod_id, game_version="1.21.1", loader="fabric"): """Fetch latest compatible version for a mod. Returns (version_id, version_number, date) or None.""" # Use API: /project/{id}/version try: req = urllib.request.Request(f"{API}/project/{mod_id}/version", headers={"User-Agent": "FKRPG-audit/1.0"}) with urllib.request.urlopen(req, timeout=10) as r: data = json.loads(r.read()) # filter by game_version + loader, newest first (API is newest first) for v in data: gvs = v.get("game_versions", []) loaders = v.get("loaders", []) if game_version not in gvs: continue if loader not in loaders and "quilt" not in loaders: # fabric mods often list fabric; be strict if loader not in loaders: continue return v.get("id"), v.get("version_number") or v.get("name"), v.get("date_published") # fallback: first entry regardless if data: v = data[0] return v.get("id"), v.get("version_number") or v.get("name"), v.get("date_published") except Exception as e: return None, None, str(e) return None, None, None def classify_counts(mods, policy): c = Counter() for slug in mods: pol = policy.get(slug, {}).get("policy", "UNKNOWN") c[pol] += 1 return c def main(): ap = argparse.ArgumentParser(description="FKRPG audit — Packwiz source of truth") ap.add_argument("--pack", default=SERVER_PACK, help="pack dir") ap.add_argument("--json", action="store_true", help="output JSON") ap.add_argument("--check-outdated", action="store_true", help="query Modrinth for outdated optimizations (network)") ap.add_argument("--quick", action="store_true", help="skip network checks") ap.add_argument("--write-manifest", action="store_true", help="update optimization-manifest.json generated_at/last_audit") args = ap.parse_args() pack_dir = args.pack mods, pack_toml, index = scan_pack(pack_dir) policy = load_policy() manifest = load_manifest() # pack info pack_info = { "minecraft": pack_toml.get("versions", {}).get("minecraft", "?"), "fabric_loader": pack_toml.get("versions", {}).get("fabric", "?"), "pack_version": pack_toml.get("version", "?"), "pack_format": pack_toml.get("pack_format", "?"), "index_hash": pack_toml.get("index", {}).get("hash", "?"), "mod_count": len(mods), } # classification by_policy = defaultdict(list) for slug in mods: pol = policy.get(slug, {}).get("policy", "UNKNOWN") by_policy[pol].append(slug) # index integrity missing, extra = check_index(pack_dir, mods, index) hash_errors = check_hashes(mods) # local drift local_jars, local_err = get_local_mods() pack_filenames = {m["filename"] for m in mods.values() if m["filename"]} drift = {} if local_jars is None: drift["local"] = {"status": "not found", "detail": local_err} else: if isinstance(local_jars, set): missing_local = sorted(pack_filenames - local_jars) extra_local = sorted(local_jars - pack_filenames) drift["local"] = { "status": "drift" if (missing_local or extra_local) else "in-sync", "pack_mods": len(pack_filenames), "local_mods": len(local_jars), "missing_in_local": missing_local[:20], "extra_in_local": extra_local[:20], "missing_count": len(missing_local), "extra_count": len(extra_local), } # also check index hash vs manifest idx_path = os.path.join(pack_dir, "index.toml") if os.path.exists(idx_path): drift["local"]["index_sha256"] = sha256_file(idx_path)[:16] # production hint prod_info = get_production_hint() drift["production_hint"] = prod_info # pack revision drift (compare pack.toml hash) # Manifest stores index_hash; compare to current pack_toml hash manifest_hash = manifest.get("pack", {}).get("index_hash", "?") current_hash = pack_toml.get("index", {}).get("hash", "?") drift["packwiz_revision"] = { "manifest_hash": manifest_hash, "current_pack_toml_hash": current_hash, "in_sync": manifest_hash == current_hash, } # outdated detection (only optimization + worldgen) outdated = [] review_required = [] safe_updates = [] if args.check_outdated and not args.quick: for slug, m in mods.items(): pol = policy.get(slug, {}).get("policy", "") if pol not in ("OPTIMIZATION", "WORLDGEN"): continue mod_id = m["mod_id"] if not mod_id: continue latest_id, latest_ver, date = fetch_modrinth_latest(mod_id) if not latest_id: continue if latest_id != m["version_id"]: entry = { "slug": slug, "policy": pol, "current": m["filename"], "current_version_id": m["version_id"], "latest_version_id": latest_id, "latest_version_number": latest_ver, } if pol == "WORLDGEN": review_required.append(entry) else: outdated.append(entry) if policy.get(slug, {}).get("automatic_updates") is False and pol == "OPTIMIZATION": # still safe to propose, but needs test safe_updates.append(entry) else: # Use static known outdated from manifest notes (offline) # Flag known drift: continents wrong version, lithium old etc. pass # overall status errors = [] if missing: errors.append(f"index missing {len(missing)} pw.toml") if extra: errors.append(f"index extra {len(extra)} entries") if hash_errors: errors.append(f"hash errors {len(hash_errors)}") if drift.get("local", {}).get("status") == "drift": errors.append("LOCAL drift detected") if not drift["packwiz_revision"]["in_sync"]: errors.append("manifest index_hash out of sync — run --write-manifest or packwiz refresh") result = { "pack": pack_info, "classification": {k: len(v) for k, v in by_policy.items()}, "by_policy": {k: sorted(v) for k, v in by_policy.items()}, "index_integrity": {"missing": missing, "extra": extra, "ok": not missing and not extra}, "hash_errors": hash_errors, "drift": drift, "outdated_optimization": outdated, "review_required_worldgen": review_required, "errors": errors, "frozen_count": len(by_policy.get("FROZEN", [])), "local_path": LOCAL_SERVER, "pack_dir": pack_dir, } if args.write_manifest and manifest: import datetime manifest["pack"]["generated_at"] = datetime.datetime.utcnow().strftime("%Y-%m-%dT%H:%M:%SZ") manifest["last_audit"] = manifest["pack"]["generated_at"] manifest["pack"]["index_hash"] = current_hash # update counts with open(MANIFEST_PATH, "w") as f: json.dump(manifest, f, indent=2, sort_keys=False) f.write("\n") print(f"Updated {MANIFEST_PATH} last_audit={manifest['last_audit']}") if args.json: print(json.dumps(result, indent=2)) sys.exit(1 if errors else 0) # Fish-friendly text print(f"=== FKRPG AUDIT — Packwiz is source of truth ===") print(f"Pack: {pack_info['minecraft']} / Fabric {pack_info['fabric_loader']} / pack {pack_info['pack_version']} ({pack_info['pack_format']})") print(f"Mods indexed: {pack_info['mod_count']} index hash: {pack_info['index_hash'][:16]}...") print(f"Classification: " + ", ".join(f"{k}={len(v)}" for k,v in sorted(by_policy.items()))) print(f" FROZEN={len(by_policy.get('FROZEN',[]))} OPTIMIZATION={len(by_policy.get('OPTIMIZATION',[]))} WORLDGEN={len(by_policy.get('WORLDGEN',[]))} LIBRARY={len(by_policy.get('LIBRARY',[]))} GAMEPLAY={len(by_policy.get('GAMEPLAY',[]))}") if missing or extra: print(f"\n[INDEX] missing={missing} extra={extra}") else: print(f"\n[INDEX] OK — all {len(mods)} pw.toml in index.toml") if hash_errors: print(f"[HASH] ERRORS: {hash_errors}") else: print(f"[HASH] OK — all downloads have sha512 + https") # drift print(f"\n[DRIFT]") loc = drift.get("local", {}) if loc.get("status") == "not found": print(f" LOCAL: not found at {LOCAL_SERVER} ({loc.get('detail')})") elif loc.get("status") == "drift": print(f" DRIFT DETECTED") print(f" Packwiz: {pack_info['mod_count']} mods hash {current_hash[:16]}...") print(f" Local: {loc['local_mods']} jars missing_in_local={loc['missing_count']} extra_in_local={loc['extra_count']}") if loc["missing_in_local"]: print(f" missing: {loc['missing_in_local'][:5]}") if loc["extra_in_local"]: print(f" extra: {loc['extra_in_local'][:5]}") print(f" Action: Local is outdated/drifted — reinstall from Packwiz export (packwiz refresh + export mrpack + install)") elif loc.get("status") == "in-sync": print(f" LOCAL: in-sync ({loc['local_mods']} jars)") else: print(f" LOCAL: {loc}") rev = drift["packwiz_revision"] if not rev["in_sync"]: print(f" PACKWIZ REVISION: manifest {rev['manifest_hash'][:16]}... != current {rev['current_pack_toml_hash'][:16]}... — run audit --write-manifest after packwiz refresh") else: print(f" PACKWIZ REVISION: in-sync {rev['current_pack_toml_hash'][:16]}...") if prod_info: print(f" PRODUCTION HINT: {json.dumps(prod_info, indent=4)}") else: print(f" PRODUCTION: no local data/ hint — check VPS via docker compose (file-mount /modpacks/pack.mrpack)") if outdated or review_required: print(f"\n[OUTDATED]") for e in outdated: print(f" SAFE UPDATE (test before prod): {e['slug']} {e['current']} → {e['latest_version_number']} ({e['latest_version_id'][:8]})") for e in review_required: print(f" REVIEW REQUIRED (worldgen, explicit approval): {e['slug']} {e['current']} → {e['latest_version_number']} ({e['latest_version_id'][:8]}) — BLOCKED until approved") else: if args.check_outdated: print(f"\n[OUTDATED] all optimization/worldgen up-to-date (or offline)") else: print(f"\n[OUTDATED] skipped (use --check-outdated for Modrinth query)") # immutable reminder print(f"\n[POLICY] FROZEN RPG/magic/combat: {len(by_policy.get('FROZEN',[]))} mods — never auto-update") if any(pol == "FROZEN" for pol in [policy.get(s,{}).get("policy") for s in outdated]): print(f" BLOCKED — optimization would touch FROZEN stack (not proposed)") if errors: print(f"\n[AUDIT FAIL] {errors}") sys.exit(1) else: print(f"\n[AUDIT PASS]") if __name__ == "__main__": main()