All checks were successful
Build & Release Pack / build (push) Successful in 1m24s
- server_pack: bump 1.0 → 1.0.1 (patch, optimization-only) - Lithium 0.15.0→0.15.4 (gvQqBUqZ/N08Z8wog) - FerriteCore 7.0.2-hotfix→7.0.3 (uXXizFIs/sOzRw3CG) - C2ME 0.3.0+alpha.0.364→0.4.0-alpha.0.27 (VSNURh3q/gRm1ZAvc) fixes drift, async gen for Tectonic 1.28 - ChunkSmith 3.13.0→3.14.0 (4BeAEBIb/iBSiBgt7) - Add Spark 1.10.109 (l6YH9Als/cALUj9l1) server-only for profiling - Tune chunksmith/config.json: targetMspt 45, tickBudget 10, heap 75, dispatch 100, settle 20 - Tune c2me-notes.txt + modernfix-mixins.properties with documented safe defaults - Add optimization-manifest.json + mod-version-policy.json (FROZEN 31/OPT 11) + audit/drift/benchmark tools + baseline + report - Packwiz is source of truth; client_pack unchanged (server-only Spark) - Validate: packwiz refresh OK 144 mods, index adccf3bb..., manifest in-sync - Drift: LOCAL test still 142 vs 144 (needs reinstall from Packwiz export); PACKWIZ internal OK
406 lines
17 KiB
Python
Executable file
406 lines
17 KiB
Python
Executable file
#!/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()
|