Release armiger-ultra v2.1.0 (BEP-44/46 Byzantine-resilient RSI, multi-instance transfer proof, UDP wire harness, 45/45 red-team hardened)
efa3810 verified | """Multi-instance proof: closed-loop RSI telemetry -> aggregator -> BEP-46 update -> receiver hot-reload / rollback. | |
| Every instance is a real `ArmigerUltraPipeline` loaded from this directory (fp32 ONNX backbone, the full | |
| 929,408-key zero-cue catalog, `swarm_adapter.pt`). Nothing is mocked: telemetry is hybrid X25519 + ML-KEM-768 | |
| encrypted, manifests carry real Ed25519 + WOTS+-SHAKE256 signatures, and the receiver's decisions come from its | |
| unchanged `classify` path. | |
| Set ARMIGER_PROOF_RESULTS=/path/results.json to dump the measured numbers used in | |
| PROOF_OF_DISTRIBUTED_KNOWLEDGE_TRANSFER.md. | |
| """ | |
| from __future__ import annotations | |
| import dataclasses | |
| import hashlib | |
| import json | |
| import math | |
| import os | |
| from pathlib import Path | |
| import random | |
| import sys | |
| import time | |
| from typing import Any, Dict, List, Sequence | |
| import numpy as np | |
| import pytest | |
| _THIS_DIR = Path(__file__).resolve().parent | |
| if str(_THIS_DIR) not in sys.path: | |
| sys.path.insert(0, str(_THIS_DIR)) | |
| from armiger_ultra_pipeline import ( # noqa: E402 | |
| ArmigerUltraPipeline, | |
| PeerAttestation, | |
| RSIUpdateAggregator, | |
| WOTSPlusSHAKE256, | |
| encode_safetensors, | |
| generate_federation_signer, | |
| ) | |
| EPOCH = "2026-10" | |
| RESULTS: Dict[str, Any] = {} | |
| PEER_ASNS = [64512, 64513, 64514, 64515, 64516, 64517] # RFC 6996 private-use ASNs (synthetic peers) | |
| # Release-name variants: contributors each grabbed a different release; the receiver sees a name nobody contributed. | |
| CONTRIB_VARIANTS = [ | |
| "{T} ({Y}) [1080p]", | |
| "{D}.{Y}.720p.WEB-DL.H264", | |
| "{T} {Y} 1080p WEBRip x265", | |
| "{D}.{Y}.2160p.WEB.H265", | |
| "{T} {Y} 1080p WEB-DL", | |
| "{D}.{Y}.1080p.WEB.h264", | |
| ] | |
| RECEIVER_VARIANT = "{T} ({Y}) [720p] [WEB-DL]" | |
| CONTROL_RECEIVER_VARIANT = "{D}.{Y}.720p.WEB-DL.H264" | |
| TARGETS = ["Hollow Meridian", "The Quiet Orchard", "Low Tide at Carrow", "Northern Relay", "Kettle Point"] | |
| CONTROL = "Ninefold" # verified TV, but the receiver's backbone is confidently (>=0.88) wrong on its release name | |
| DECOY = "Saltmarsh" # seen by one contributor only -> must not reach the update | |
| def _dump_results(): | |
| yield | |
| out = os.environ.get("ARMIGER_PROOF_RESULTS") | |
| if out: | |
| Path(out).write_text(json.dumps(RESULTS, indent=2, sort_keys=True, default=str)) | |
| def federation() -> Dict[str, Any]: | |
| return { | |
| "signers": [generate_federation_signer() for _ in range(3)], # 2-of-3 threshold | |
| "peers": [{"asn": asn, "signer": generate_federation_signer()} for asn in PEER_ASNS], | |
| } | |
| def _configure(pipe: ArmigerUltraPipeline, fed: Dict[str, Any]) -> ArmigerUltraPipeline: | |
| pipe.configure_dht_update_trust( | |
| trusted_signers={s[1]: f"federation-{i}" for i, s in enumerate(fed["signers"])}, | |
| trusted_pq_signers={s[3]: f"federation-{i}" for i, s in enumerate(fed["signers"])}, | |
| threshold_m=2, | |
| trusted_attestors={p["signer"][1]: {"asn": p["asn"], "pq_pubkey_hex": p["signer"][3]} for p in fed["peers"]}, | |
| min_distinct_asns=5, | |
| require_pq=True, | |
| ) | |
| return pipe | |
| def make_instance(federation): | |
| created: List[ArmigerUltraPipeline] = [] | |
| def _make() -> ArmigerUltraPipeline: | |
| pipe = _configure(ArmigerUltraPipeline.from_Directory(_THIS_DIR, preload_catalog=True), federation) | |
| created.append(pipe) | |
| return pipe | |
| yield _make | |
| for pipe in created: | |
| pipe.close() | |
| # --------------------------------------------------------------------------------------------- | |
| # Physical-import fixtures (*Arr webhooks, qBittorrent completion, ffprobe stream inspection) | |
| # --------------------------------------------------------------------------------------------- | |
| def _name(title: str, variant: str, year: int = 2026) -> str: | |
| return variant.format(T=title, D=title.replace(" ", "."), Y=year) | |
| def _tv_pack_files(title: str, year: int = 2026) -> List[Dict[str, Any]]: | |
| return [{"path": f"{title} ({year})/0{j}.mkv", "size": 2_300_000_000} for j in (1, 2, 3)] | |
| def _ffprobe(size: int, duration: float = 3120.0) -> Dict[str, Any]: | |
| return { | |
| "streams": [ | |
| {"codec_type": "video", "codec_name": "h264", "width": 1920, "height": 1080}, | |
| {"codec_type": "audio", "codec_name": "eac3", "channels": 6, "tags": {"language": "eng"}}, | |
| {"codec_type": "subtitle", "codec_name": "subrip", "tags": {"language": "eng"}}, | |
| ], | |
| "format": {"format_name": "matroska,webm", "duration": str(duration), "size": str(size)}, | |
| } | |
| def _series_webhook(series_title: str, catalog_id: int, release_title: str, files: Sequence[Dict[str, Any]]) -> Dict[str, Any]: | |
| size = int(sum(f["size"] for f in files)) | |
| return { | |
| "eventType": "Download", | |
| "series": {"title": series_title, "seriesCatalogId": catalog_id}, | |
| "episodes": [{"seasonNumber": 1, "episodeNumber": j + 1} for j in range(len(files))], | |
| "release": {"releaseTitle": release_title, "size": size}, | |
| "episodeFile": { | |
| "relativePath": files[0]["path"], "size": size, "duration": 3120, | |
| "mediaInfo": {"videoCodec": "h264", "audioFormat": "eac3", "audioChannels": 5.1, | |
| "audioLanguages": "eng", "subtitles": "eng"}, | |
| }, | |
| } | |
| def _movie_webhook(title: str, year: int, catalog_id: int, release_title: str, files: Sequence[Dict[str, Any]]) -> Dict[str, Any]: | |
| size = int(sum(f["size"] for f in files)) | |
| return { | |
| "eventType": "Download", | |
| "movie": {"title": title, "year": year, "movieCatalogId": catalog_id}, | |
| "release": {"releaseTitle": release_title, "size": size}, | |
| "movieFile": { | |
| "relativePath": files[0]["path"], "size": size, "duration": 6400, | |
| "mediaInfo": {"videoCodec": "h264", "audioFormat": "eac3", "audioChannels": 5.1, | |
| "audioLanguages": "eng", "subtitles": "eng"}, | |
| }, | |
| } | |
| def _qbit_completion(name: str, size: int) -> Dict[str, Any]: | |
| return {"hash": hashlib.sha1(name.encode()).hexdigest(), "name": name, "size": size, | |
| "progress": 1.0, "state": "stalledUP", "content_path": f"/downloads/complete/{name}"} | |
| def _export(pipe, agg, payload, *, adapter, row, ffprobe=None): | |
| ok, reason, bundle = pipe.export_rsi_telemetry_bundle( | |
| payload, adapter=adapter, torrent_row=row, ffprobe_output=ffprobe, epoch=EPOCH, | |
| aggregator_x25519_pub=agg.x25519_pub, aggregator_mlkem_ek=agg.mlkem_ek, | |
| ) | |
| assert ok, reason | |
| assert bundle.is_post_quantum and len(bytes.fromhex(bundle.ephemeral_pubkey_hex)) == 32 + 1088 | |
| assert bundle.cipher_suite == "X25519-MLKEM768-ChaCha20Poly1305-SHA3" | |
| return bundle | |
| def _attest_all(peers, fed, payload): | |
| attestations, reviews = [], [] | |
| for pipe, peer in zip(peers, fed["peers"]): | |
| att, review = pipe.attest_dht_update(payload, signer=peer["signer"], asn=peer["asn"], peer_id=f"peer-asn{peer['asn']}") | |
| reviews.append(review) | |
| if att is not None: | |
| attestations.append(att) | |
| return attestations, reviews | |
| def _forged_attestations(fed, payload, peers=None) -> List[PeerAttestation]: | |
| """Attacker model for Test 3: the attestor KEYS are compromised, so attestations are signed without review.""" | |
| target = hashlib.sha256(payload).hexdigest() | |
| out = [] | |
| for peer in (peers or fed["peers"]): | |
| ed_priv, ed_pub, pq_priv, pq_pub = peer["signer"] | |
| unsigned = PeerAttestation(peer_id=f"asn{peer['asn']}", asn=peer["asn"], peer_pubkey_hex=ed_pub, | |
| timestamp=int(time.time()), signature_hex="") | |
| msg = unsigned.canonical_bytes(target) | |
| out.append(dataclasses.replace(unsigned, signature_hex=ed_priv.sign(msg).hex(), pq_pubkey_hex=pq_pub, | |
| pq_signature_hex=WOTSPlusSHAKE256.sign(pq_priv, msg).hex())) | |
| return out | |
| def _cat_probs(pred) -> Dict[str, float]: | |
| p = pred.heads["category"]["probs"] | |
| return {"movie": round(p["movie"], 4), "tv_show": round(p["tv_show"], 4)} | |
| # ============================================================================================= | |
| # Test 1: Instance A learns zero-cue titles -> DHT -> Instance B improves accuracy | |
| # ============================================================================================= | |
| def test_1_zero_cue_knowledge_transfers_from_instance_a_to_instance_b(make_instance, federation): | |
| instance_b = make_instance() | |
| b_rows = {t: {"name": _name(t, RECEIVER_VARIANT), "size": 6_900_000_000, "files": _tv_pack_files(t)} for t in TARGETS} | |
| control_row = {"name": _name(CONTROL, CONTROL_RECEIVER_VARIANT), "size": 6_900_000_000, "files": _tv_pack_files(CONTROL)} | |
| # --- Before: B cannot disambiguate the brand-new bare-title 2026 releases -------------------------- | |
| before = {t: instance_b.classify(r) for t, r in b_rows.items()} | |
| for t, pred in before.items(): | |
| src = pred.plugin_trace["plugin_3_zero_cue_catalog"]["resolution_source"] | |
| assert pred.category != "tv_show", f"precondition: B must start wrong on {t}" | |
| assert src == "neural_backbone", f"precondition: {t} must be absent from the shipped catalog" | |
| assert max(_cat_probs(pred).values()) < 0.88, f"precondition: {t} must be inside the catalog gate" | |
| control_before = instance_b.classify(control_row) | |
| assert control_before.category == "movie" and max(_cat_probs(control_before).values()) >= 0.88 | |
| # --- Instance A + 5 peers (distinct ASNs) import the releases through SeriesWebhookAdapter + ffprobe ------------ | |
| agg = RSIUpdateAggregator(epoch=EPOCH) | |
| instance_a = make_instance() | |
| peers = [instance_a] + [make_instance() for _ in range(5)] | |
| bundles = [] | |
| for i, pipe in enumerate(peers): | |
| for k, title in enumerate(TARGETS + [CONTROL]): | |
| name = _name(title, CONTRIB_VARIANTS[i]) | |
| files = _tv_pack_files(title) | |
| row = {"name": name, "size": sum(f["size"] for f in files), "files": files} | |
| bundles.append(_export(pipe, agg, _series_webhook(title, 910_001 + k, name, files), | |
| adapter="series_webhook", row=row, ffprobe=_ffprobe(row["size"]))) | |
| decoy_files = _tv_pack_files(DECOY) | |
| decoy_name = _name(DECOY, CONTRIB_VARIANTS[0]) | |
| bundles.append(_export(instance_a, agg, _series_webhook(DECOY, 919_999, decoy_name, decoy_files), | |
| adapter="series_webhook", row={"name": decoy_name, "size": 6_900_000_000, "files": decoy_files}, | |
| ffprobe=_ffprobe(6_900_000_000))) | |
| # An unplayable "import" is refused before any telemetry exists | |
| bad_ok, bad_reason, bad_bundle = instance_a.export_rsi_telemetry_bundle( | |
| _series_webhook("Kettle Point", 910_005, _name("Kettle Point", CONTRIB_VARIANTS[0]), _tv_pack_files("Kettle Point")), | |
| adapter="series_webhook", ffprobe_output={"streams": [], "format": {}}, epoch=EPOCH, | |
| aggregator_x25519_pub=agg.x25519_pub, aggregator_mlkem_ek=agg.mlkem_ek) | |
| assert not bad_ok and bad_bundle is None and "invalid_unplayable_media_stream" in bad_reason | |
| # --- Aggregator: decrypt/authenticate, refuse replay + tamper, unanimous multi-contributor consensus --- | |
| tampered = dataclasses.replace(bundles[0], ciphertext_hex=bundles[0].ciphertext_hex[:-2] + ("00" if bundles[0].ciphertext_hex[-2:] != "00" else "01")) | |
| first = agg.ingest_bundles([tampered] + bundles) | |
| replay = agg.ingest_bundles([bundles[3]]) | |
| assert first == {"accepted": len(bundles), "rejected": 1} and replay == {"accepted": 0, "rejected": 1} | |
| assert any(r.startswith("aead_or_json_rejected") for r in agg.rejected) and "duplicate_bundle_replay" in agg.rejected | |
| plaintext = json.dumps(agg.observations).lower() | |
| for title in TARGETS + [CONTROL, DECOY]: | |
| assert title.lower() not in plaintext and title.lower().replace(" ", ".") not in plaintext | |
| delta, zc_stats = agg.build_zero_cue_delta(min_support=5) | |
| assert zc_stats == {"keys_seen": 7, "accepted": 6, "conflicting": 0, "insufficient_support": 1} | |
| payload = agg.build_update_payload(zero_cue_catalog_delta=delta) | |
| attestations, reviews = _attest_all(peers, federation, payload) | |
| assert [r.status for r in reviews] == ["review_passed"] * 6 | |
| assert len(attestations) == 6 | |
| manifest = RSIUpdateAggregator.sign_manifest(payload, seq=1, signers=federation["signers"][:2], | |
| version="zero-cue-2026.10.18", min_asn_quorum=5) | |
| # --- Instance B verifies, stages, canary-gates and hot-reloads ----------------------------------- | |
| fp_before = instance_b.dht_state_fingerprint() | |
| result = instance_b.stage_and_apply_dht_update(manifest, payload, attestations=attestations, expected_prev_seq=0) | |
| assert result.applied and result.status == "committed", result.reason | |
| assert (result.seq_before, result.seq_after) == (0, 1) | |
| assert all(s["ok"] for s in result.verification_trace), result.verification_trace | |
| assert result.regressions == [] and result.state_fingerprint_after != fp_before | |
| after = {t: instance_b.classify(r) for t, r in b_rows.items()} | |
| correct = sum(p.category == "tv_show" for p in after.values()) | |
| assert correct == len(TARGETS) # 5/5 = 100% | |
| for pred in after.values(): | |
| assert pred.plugin_trace["plugin_3_zero_cue_catalog"]["resolution_source"].startswith("zero_cue_") | |
| control_after = instance_b.classify(control_row) | |
| assert control_after.category == "movie" # confident backbone predictions are never overridden by catalog data | |
| assert control_after.plugin_trace["plugin_3_zero_cue_catalog"]["resolution_source"] == "neural_backbone" | |
| # --- Replay of the same signed update is refused and changes nothing ---------------------------- | |
| fp_committed = instance_b.dht_state_fingerprint() | |
| replayed = instance_b.stage_and_apply_dht_update(manifest, payload, attestations=attestations) | |
| assert replayed.status == "rejected_verification" and "stale_or_replayed_seq" in replayed.reason | |
| assert instance_b.dht_state_fingerprint() == fp_committed | |
| RESULTS["test_1"] = { | |
| "receiver_release_names": {t: r["name"] for t, r in b_rows.items()}, | |
| "before": {t: {"category": p.category, **_cat_probs(p)} for t, p in before.items()}, | |
| "after": {t: {"category": p.category, **_cat_probs(p), | |
| "source": p.plugin_trace["plugin_3_zero_cue_catalog"]["resolution_source"]} for t, p in after.items()}, | |
| "accuracy_before": f"{sum(p.category == 'tv_show' for p in before.values())}/5", | |
| "accuracy_after": f"{correct}/5", | |
| "control": {"name": control_row["name"], "before": {"category": control_before.category, **_cat_probs(control_before)}, | |
| "after": {"category": control_after.category, **_cat_probs(control_after)}}, | |
| "bundles_exported": len(bundles), "aggregator_rejections": list(agg.rejected), "zero_cue_consensus": zc_stats, | |
| "payload_sha256": hashlib.sha256(payload).hexdigest(), "payload_bytes": len(payload), | |
| "manifest": {"seq": manifest.seq, "infohash": manifest.infohash, "ed25519_signers": len(manifest.signatures), | |
| "wots_signers": len(manifest.pq_signatures), | |
| "wots_sig_bytes": len(bytes.fromhex(next(iter(manifest.pq_signatures.values()))))}, | |
| "peer_reviews": [{"status": r.status, "elapsed_ms": r.elapsed_ms, | |
| "ledger": next((s["detail"] for s in r.verification_trace if s["step"] == "ledger_confirmation"), None)} for r in reviews], | |
| "receiver_trace": result.verification_trace, | |
| "receiver_canary_baseline": result.baseline_canary, "receiver_canary_candidate": result.candidate_canary, | |
| "receiver_elapsed_ms": result.elapsed_ms, | |
| "replay": {"status": replayed.status, "reason": replayed.reason}, | |
| } | |
| # ============================================================================================= | |
| # Test 2: distributed calibration + SwarmResidualAdapter weight-delta transfer | |
| # ============================================================================================= | |
| _A = ["Amber", "Cobalt", "Willow", "Granite", "Thistle", "Harrow", "Juniper", "Marsh", "Copper", "Linden", "Briar", "Oriel"] | |
| _B = ["Reach", "Hollow", "Ferry", "Station", "Lantern", "Crossing", "Ledger", "Acre", "Signal", "Tide", "Ridge", "Orchard"] | |
| _G = ["NTb", "FLUX", "SiGMA", "EDITH", "CMRG"] | |
| def _clean_swarm(rng: random.Random, i: int) -> Dict[str, Any]: | |
| if i % 4 == 0: # legitimately dead swarm with real completion history | |
| return {"tracker_known": True, "tracker_seeders": 0, "tracker_leechers": rng.randint(10, 40), | |
| "tracker_completed": rng.randint(100, 3000), "prev_seeders": rng.randint(0, 9)} | |
| return {"tracker_known": True, "tracker_seeders": rng.randint(3, 500), "tracker_leechers": rng.randint(0, 80), | |
| "tracker_completed": rng.randint(20, 6000)} | |
| def _trap_swarm(rng: random.Random, i: int) -> Dict[str, Any]: | |
| if i % 2 == 0: # fake seeders that never complete | |
| return {"tracker_known": True, "tracker_seeders": rng.randint(50, 600), "tracker_leechers": rng.randint(0, 30), | |
| "tracker_completed": 0} | |
| return {"tracker_known": True, "tracker_seeders": 0, "tracker_leechers": rng.randint(12, 300), | |
| "tracker_completed": rng.randint(0, 2), "prev_seeders": rng.randint(10, 80)} | |
| def _media_pool(seed: int, n: int) -> List[Dict[str, Any]]: | |
| """Verified library imports: bare-title miniseries packs, BluRay movies, and SxxEyy episodes.""" | |
| rng = random.Random(seed) | |
| out = [] | |
| for i in range(n): | |
| title = f"{rng.choice(_A)} {rng.choice(_B)}" | |
| kind = i % 3 | |
| if kind == 0: | |
| name = f"{title} (2026) [1080p]" | |
| files = [{"path": f"{title} (2026)/0{j}.mkv", "size": 2_000_000_000} for j in (1, 2, 3)] | |
| label = "tv_show" | |
| elif kind == 1: | |
| year = rng.randint(1995, 2025) | |
| name = f"{title.replace(' ', '.')}.{year}.1080p.BluRay.x264-{rng.choice(_G)}" | |
| files = [{"path": f"{name}.mkv", "size": 9_000_000_000}] | |
| label = "movie" | |
| else: | |
| name = f"{title.replace(' ', '.')}.S0{rng.randint(1, 5)}E0{rng.randint(1, 9)}.1080p.WEB.h264-{rng.choice(_G)}" | |
| files = [{"path": f"{name}.mkv", "size": 1_500_000_000}] | |
| label = "tv_show" | |
| out.append({"title": title, "name": name, "files": files, "label": label, | |
| "swarm": _clean_swarm(rng, i), "year": 2026 if kind == 0 else None}) | |
| return out | |
| def _junk_pool(seed: int, n: int) -> List[Dict[str, Any]]: | |
| """Completed downloads that the physical verifier flags as masquerading executables, with trap swarms.""" | |
| rng = random.Random(seed) | |
| out = [] | |
| for i in range(n): | |
| title = f"{rng.choice(_A)}.{rng.choice(_B)}" | |
| ext = ("mkv.exe", "mp4.scr", "avi.exe")[i % 3] | |
| name = f"{title}.2026.1080p.WEB-DL.{ext}" | |
| out.append({"name": name, "files": [{"path": name, "size": 2_400_000}], "swarm": _trap_swarm(rng, i)}) | |
| return out | |
| def _import_media(pipe, agg, item, catalog_id): | |
| size = int(sum(f["size"] for f in item["files"])) | |
| row = {"name": item["name"], "size": size, "files": item["files"], "swarm": item["swarm"]} | |
| if item["label"] == "movie": | |
| hook = _movie_webhook(item["title"], 2000, catalog_id, item["name"], item["files"]) | |
| return _export(pipe, agg, hook, adapter="movie_webhook", row=row, ffprobe=_ffprobe(size, 6400.0)) | |
| hook = _series_webhook(item["title"], catalog_id, item["name"], item["files"]) | |
| return _export(pipe, agg, hook, adapter="series_webhook", row=row, ffprobe=_ffprobe(size)) | |
| def _import_junk(pipe, agg, item): | |
| row = {"name": item["name"], "size": 2_400_000, "files": item["files"], "swarm": item["swarm"]} | |
| return _export(pipe, agg, _qbit_completion(item["name"], 2_400_000), adapter="qbittorrent", row=row) | |
| def _adapter_trap_scores(pipe, rows) -> List[float]: | |
| """The SwarmResidualAdapter's own junk-margin contribution, computed exactly as in `evaluate_swarm`.""" | |
| import torch | |
| from swarm_priors import extract_way2_source_features, extract_way3_anomaly_features | |
| logits = pipe._run_neural_logits(rows) | |
| scores = [] | |
| with torch.no_grad(): | |
| for i, r in enumerate(rows): | |
| pj = float(pipe.calibrator.calibrate_head("junk", logits["junk"][i].tolist())[1]) | |
| pj = min(max(pj, 1e-6), 1 - 1e-6) | |
| zl = math.log(pj / (1 - pj)) | |
| out = pipe.swarm_adapter.forward( | |
| torch.zeros(1, 384), torch.zeros(1, 24), {"junk": torch.tensor([[0.0, zl]])}, | |
| torch.tensor([list(extract_way2_source_features(r, swarm=r["swarm"]))], dtype=torch.float32), | |
| torch.tensor([list(extract_way3_anomaly_features(r, swarm=r["swarm"]))], dtype=torch.float32), | |
| ) | |
| scores.append(float(out["junk"][0, 1] - out["junk"][0, 0]) - zl) | |
| return scores | |
| def _auroc(scores: Sequence[float], labels: Sequence[int]) -> float: | |
| pos = [s for s, y in zip(scores, labels) if y] | |
| neg = [s for s, y in zip(scores, labels) if not y] | |
| return sum((p > n) + 0.5 * (p == n) for p in pos for n in neg) / (len(pos) * len(neg)) | |
| def test_2_calibration_and_swarm_adapter_delta_transfer(make_instance, federation): | |
| instance_b = make_instance() | |
| agg = RSIUpdateAggregator(epoch=EPOCH) | |
| peers = [make_instance() for _ in range(6)] # instance A + 5 peers | |
| bundles = [] | |
| for i, pipe in enumerate(peers): | |
| for k, item in enumerate(_media_pool(100 + i, 15)): | |
| bundles.append(_import_media(pipe, agg, item, 920_000 + 100 * i + k)) | |
| for item in _junk_pool(200 + i, 6): | |
| bundles.append(_import_junk(pipe, agg, item)) | |
| assert agg.ingest_bundles(bundles) == {"accepted": 126, "rejected": 0} | |
| cal_delta, cal_stats = agg.fit_calibration_delta(heads=("category",), min_samples=20) | |
| assert "category" in cal_delta["heads"], cal_stats | |
| adapter_delta, ad_stats = agg.train_swarm_adapter_delta(instance_b.swarm_adapter, steps=300, seed=0) | |
| assert adapter_delta is not None, ad_stats | |
| assert ad_stats["bce_after"] < ad_stats["bce_before"] | |
| payload = agg.build_update_payload(calibration_delta=cal_delta, swarm_adapter_delta=adapter_delta) | |
| # Held-out evaluation data that NO instance contributed | |
| held_media = _media_pool(999, 60) | |
| held_rows = [{"name": it["name"], "size": sum(f["size"] for f in it["files"]), "files": it["files"]} for it in held_media] | |
| held_labels = [it["label"] for it in held_media] | |
| trap_rng = random.Random(4242) | |
| swarm_eval_rows, swarm_eval_y = [], [] | |
| for j, it in enumerate(_junk_pool(777, 20)): | |
| swarm_eval_rows.append({"name": it["name"], "size": 2_400_000, "files": it["files"], "swarm": it["swarm"]}) | |
| swarm_eval_y.append(1) | |
| for j, it in enumerate(_media_pool(778, 20)): | |
| swarm_eval_rows.append({"name": it["name"], "size": sum(f["size"] for f in it["files"]), "files": it["files"], | |
| "swarm": _clean_swarm(trap_rng, j)}) | |
| swarm_eval_y.append(0) | |
| # Clean-named releases sitting on trap swarms (oracle-labelled; cannot be physically verified) | |
| clean_named_traps = [] | |
| for j, it in enumerate(_media_pool(779, 20)): | |
| clean_named_traps.append({"name": it["name"], "size": sum(f["size"] for f in it["files"]), "files": it["files"], | |
| "swarm": _trap_swarm(trap_rng, j)}) | |
| clean_named_y = [1] * 20 + [0] * 20 | |
| clean_named_rows = clean_named_traps + swarm_eval_rows[20:] | |
| nll_before = instance_b.calibration_nll(held_rows, held_labels, head="category") | |
| auroc_before = _auroc(_adapter_trap_scores(instance_b, swarm_eval_rows), swarm_eval_y) | |
| auroc_clean_named_before = _auroc(_adapter_trap_scores(instance_b, clean_named_rows), clean_named_y) | |
| preds_before = instance_b.classify_batch(swarm_eval_rows + clean_named_traps) | |
| base_state = {k: v.detach().clone() for k, v in instance_b.swarm_adapter.state_dict().items()} | |
| attestations, reviews = _attest_all(peers, federation, payload) | |
| assert [r.status for r in reviews] == ["review_passed"] * 6, [r.reason for r in reviews] | |
| manifest = RSIUpdateAggregator.sign_manifest(payload, seq=1, signers=federation["signers"][1:], | |
| version="calib-adapter-2026.10.18", min_asn_quorum=5) | |
| result = instance_b.stage_and_apply_dht_update(manifest, payload, attestations=attestations, expected_prev_seq=0) | |
| assert result.applied and result.status == "committed", (result.reason, result.regressions) | |
| assert sorted(result.components) == ["calibration_delta", "swarm_adapter_safetensors_delta"] | |
| # Transfer fidelity: B now holds exactly base + aggregator delta, and the fitted temperature | |
| new_state = instance_b.swarm_adapter.state_dict() | |
| for name, d in adapter_delta.items(): | |
| expected = base_state[name] + __import__("torch").from_numpy(d) | |
| assert bool((new_state[name] == expected).all()), name | |
| untouched = [k for k in base_state if k not in adapter_delta] | |
| assert all(bool((new_state[k] == base_state[k]).all()) for k in untouched) | |
| assert instance_b.calibrator.heads["category"]["temperature"] == cal_delta["heads"]["category"]["temperature"] | |
| nll_after = instance_b.calibration_nll(held_rows, held_labels, head="category") | |
| auroc_after = _auroc(_adapter_trap_scores(instance_b, swarm_eval_rows), swarm_eval_y) | |
| auroc_clean_named_after = _auroc(_adapter_trap_scores(instance_b, clean_named_rows), clean_named_y) | |
| preds_after = instance_b.classify_batch(swarm_eval_rows + clean_named_traps) | |
| assert nll_after < nll_before | |
| # Measured result (not tuned): the trained junk-path delta does not reorder held-out traps. The adapter's | |
| # residual is scaled by the backbone's junk uncertainty (clamped at 0.02 for confident-clean content), so | |
| # a bounded delta cannot move the ranking. The claim tested here is transfer + non-regression only. | |
| assert auroc_after >= auroc_before and auroc_clean_named_after >= auroc_clean_named_before | |
| def _rates(preds): | |
| verified_style, clean_named = preds[:40], preds[40:] | |
| return { | |
| "masquerade_traps_junk": sum(p.junk == "junk" for p in verified_style[:20]), | |
| "clean_media_junk_false_positives": sum(p.junk == "junk" for p in verified_style[20:]), | |
| "clean_named_traps_junk": sum(p.junk == "junk" for p in clean_named), | |
| "clean_named_traps_honeypot": sum(p.honeypot_trap["is_honeypot_trap"] for p in clean_named), | |
| "categories_changed": None, | |
| } | |
| rates_before, rates_after = _rates(preds_before), _rates(preds_after) | |
| flips = sum(a.category != b.category for a, b in zip(preds_before, preds_after)) | |
| rates_after["categories_changed"] = flips | |
| assert rates_after["clean_media_junk_false_positives"] <= rates_before["clean_media_junk_false_positives"] | |
| RESULTS["test_2"] = { | |
| "contributions": {"bundles": len(bundles), "calibration_profiles": cal_stats["category"]["n"], | |
| "swarm_samples": ad_stats["n_samples"], "swarm_junk_samples": ad_stats["n_junk"]}, | |
| "calibration_fit": cal_stats, "adapter_training": ad_stats, | |
| "held_out_category_nll": {"n": len(held_rows), "before": round(nll_before, 6), "after": round(nll_after, 6)}, | |
| "adapter_trap_auroc_verified_style": {"n": 40, "before": round(auroc_before, 4), "after": round(auroc_after, 4)}, | |
| "adapter_trap_auroc_clean_named": {"n": 40, "before": round(auroc_clean_named_before, 4), "after": round(auroc_clean_named_after, 4)}, | |
| "label_rates_before": rates_before, "label_rates_after": rates_after, | |
| "payload_sha256": hashlib.sha256(payload).hexdigest(), "payload_bytes": len(payload), | |
| "receiver_trace": result.verification_trace, | |
| "receiver_canary_baseline": result.baseline_canary, "receiver_canary_candidate": result.candidate_canary, | |
| "peer_reviews": [{"status": r.status, "elapsed_ms": r.elapsed_ms, | |
| "calibration_check": next((s["detail"] for s in r.verification_trace if s["step"].startswith("ledger_calibration_nll")), None)} | |
| for r in reviews], | |
| "receiver_elapsed_ms": result.elapsed_ms, | |
| } | |
| # ============================================================================================= | |
| # Test 3: poisoned updates are rejected and rolled back; predictions stay bit-identical | |
| # ============================================================================================= | |
| def _probe_rows() -> List[Dict[str, Any]]: | |
| rows = [{"name": _name(t, RECEIVER_VARIANT), "size": 6_900_000_000, "files": _tv_pack_files(t)} for t in TARGETS] | |
| for it in _junk_pool(31337, 6): | |
| rows.append({"name": it["name"], "size": 2_400_000, "files": it["files"], "swarm": it["swarm"]}) | |
| for it in _media_pool(31338, 12): | |
| rows.append({"name": it["name"], "size": sum(f["size"] for f in it["files"]), "files": it["files"], "swarm": it["swarm"]}) | |
| rows.append({"name": "Codec_Update_Required_To_Play.scr", "size": 900_000, | |
| "files": [{"path": "Codec_Update_Required_To_Play.scr", "size": 900_000}]}) | |
| rows.append({"name": "Adult_Studio.24.10.18.Scene3.XXX.1080p.MP4-ARC", "size": 1_400_000_000, | |
| "files": [{"path": "Adult_Studio.24.10.18.Scene3.XXX.1080p.MP4-ARC.mp4", "size": 1_400_000_000}]}) | |
| return rows | |
| def _poison_adapter_delta(pipe) -> Dict[str, np.ndarray]: | |
| sd = pipe.swarm_adapter.state_dict() | |
| enc = np.zeros(tuple(sd["junk_encoder.4.weight"].shape), dtype=np.float32) | |
| enc[1, :] = 60.0 # drive the junk logit up for every swarm-bearing item | |
| return { | |
| "junk_encoder.4.weight": enc, | |
| "junk_gate.bias": np.array([0.0, 20.0], dtype=np.float32), | |
| "head_scales.junk": np.array([1.25], dtype=np.float32), | |
| } | |
| def test_3_poisoned_weight_updates_roll_back_with_identical_predictions(make_instance, federation): | |
| instance_b = make_instance() | |
| probe = _probe_rows() | |
| agg = RSIUpdateAggregator(epoch=EPOCH) | |
| signers = federation["signers"] | |
| def snapshot(): | |
| return [p.to_dict() for p in instance_b.classify_batch(probe)], instance_b.dht_state_fingerprint() | |
| preds0, fp0 = snapshot() | |
| staged_cases = { | |
| "calibration_junk_threshold_0.97": agg.build_update_payload(calibration_delta={"mondrian_thresholds": {"junk": 0.97}}), | |
| "calibration_junk_temperature_6.0": agg.build_update_payload(calibration_delta={"heads": {"junk": {"temperature": 6.0}}}), | |
| "adapter_junk_flood_delta": agg.build_update_payload(swarm_adapter_delta=_poison_adapter_delta(instance_b)), | |
| } | |
| staged_results = {} | |
| for seq_offset, (case, payload) in enumerate(staged_cases.items(), start=1): | |
| manifest = RSIUpdateAggregator.sign_manifest(payload, seq=seq_offset, signers=signers[:2], | |
| version=f"poison-{case}", min_asn_quorum=5) | |
| res = instance_b.stage_and_apply_dht_update(manifest, payload, attestations=_forged_attestations(federation, payload), | |
| expected_prev_seq=0) | |
| # every cryptographic check passed: the colluding quorum is valid, only the canary stands in the way | |
| assert [s["step"] for s in res.verification_trace if not s["ok"]] == ["golden_canary_non_regression"], res.verification_trace | |
| assert res.status == "rolled_back" and not res.applied and res.regressions, case | |
| assert res.seq_after == 0 and res.state_fingerprint_after == fp0 | |
| preds, fp = snapshot() | |
| assert fp == fp0 and preds == preds0, f"{case}: predictions changed after rollback" | |
| staged_results[case] = {"reason": res.reason, "regressions": res.regressions, | |
| "candidate_canary": res.candidate_canary, "elapsed_ms": res.elapsed_ms} | |
| # Honest peers refuse to attest a poisoned payload (defence in depth before the receiver's canary) | |
| honest = make_instance() | |
| att, review = honest.attest_dht_update(staged_cases["calibration_junk_threshold_0.97"], | |
| signer=federation["peers"][0]["signer"], asn=PEER_ASNS[0]) | |
| assert att is None and review.status == "review_failed" | |
| # Rejections before staging: nothing is touched | |
| good = agg.build_update_payload(calibration_delta={"heads": {"category": {"temperature": 0.9}}}) | |
| good_atts = _forged_attestations(federation, good) | |
| m_good = RSIUpdateAggregator.sign_manifest(good, seq=1, signers=signers[:2], version="v", min_asn_quorum=5) | |
| sybils = [{"asn": 65000 + i, "signer": generate_federation_signer()} for i in range(6)] | |
| rogue = generate_federation_signer() | |
| m_one_sig = RSIUpdateAggregator.sign_manifest(good, seq=1, signers=signers[:1], version="v", min_asn_quorum=5) | |
| m_rogue_pq = dataclasses.replace(m_good, pq_signatures={rogue[3]: WOTSPlusSHAKE256.sign(rogue[2], m_good.canonical_payload()).hex(), | |
| signers[1][3]: m_good.pq_signatures[signers[1][3]]}) | |
| canary_override = json.dumps({"schema": "armiger-dht-update/v1", "epoch": EPOCH, | |
| "calibration_delta": {"heads": {"category": {"temperature": 0.9}}}, | |
| "golden_canary_override": [{"item_id": "canary_jnk_000", "junk": False}]}, | |
| sort_keys=True, separators=(",", ":")).encode() | |
| pickle_blob = b"\x80\x04\x95" + b"cos\nsystem\n" + b"\x00" * 64 | |
| pickle_st = (len(b'{"w":{"dtype":"F32","shape":[1],"data_offsets":[0,4]}}')).to_bytes(8, "little") + \ | |
| b'{"w":{"dtype":"F32","shape":[1],"data_offsets":[0,4]}}' + b"\x00\x00\x80\x04" + pickle_blob | |
| nan_adapter = encode_safetensors({"junk_gate.bias": np.array([0.0, np.nan], dtype=np.float32)}) | |
| pre_cases = { | |
| "payload_tampered_after_signing": (m_good, good.replace(b"0.9", b"0.1"), good_atts, None), | |
| "only_one_of_three_signers": (m_one_sig, good, good_atts, None), | |
| "rogue_wots_key_not_pinned": (m_rogue_pq, good, good_atts, None), | |
| "four_distinct_asns_only": (m_good, good, _forged_attestations(federation, good, federation["peers"][:4]), None), | |
| "sybil_unregistered_attestors": (m_good, good, _forged_attestations(federation, good, sybils), None), | |
| "bep44_cas_mismatch": (m_good, good, good_atts, 7), | |
| "golden_canary_override_smuggled": (RSIUpdateAggregator.sign_manifest(canary_override, seq=1, signers=signers[:2], version="v", min_asn_quorum=5), | |
| canary_override, _forged_attestations(federation, canary_override), None), | |
| "pickle_bytecode_in_safetensors": (RSIUpdateAggregator.sign_manifest(pickle_st, seq=1, signers=signers[:2], version="v", min_asn_quorum=5, bundle_format="safetensors"), | |
| pickle_st, _forged_attestations(federation, pickle_st), None), | |
| "nan_adapter_delta": (RSIUpdateAggregator.sign_manifest(nan_adapter, seq=1, signers=signers[:2], version="v", min_asn_quorum=5, bundle_format="safetensors"), | |
| nan_adapter, _forged_attestations(federation, nan_adapter), None), | |
| } | |
| pre_results = {} | |
| for case, (manifest, payload, atts, cas) in pre_cases.items(): | |
| res = instance_b.stage_and_apply_dht_update(manifest, payload, attestations=atts, expected_prev_seq=cas) | |
| assert not res.applied and res.status in ("rejected_verification", "rolled_back"), case | |
| assert res.seq_after == 0 and instance_b.dht_state_fingerprint() == fp0, case | |
| pre_results[case] = {"status": res.status, "reason": res.reason} | |
| assert pre_results["nan_adapter_delta"]["status"] == "rolled_back" # caught while staging; restored | |
| assert all(v["status"] == "rejected_verification" for k, v in pre_results.items() if k != "nan_adapter_delta") | |
| preds_final, fp_final = snapshot() | |
| assert fp_final == fp0 and preds_final == preds0 | |
| # The same receiver still accepts a legitimate update afterwards (rollback left it healthy) | |
| ok = instance_b.stage_and_apply_dht_update(m_good, good, attestations=good_atts, expected_prev_seq=0) | |
| assert ok.applied and ok.seq_after == 1, (ok.reason, ok.regressions) | |
| RESULTS["test_3"] = { | |
| "probe_rows": len(probe), "fingerprint_before": fp0, "fingerprint_after_all_rejections": fp_final, | |
| "predictions_identical": preds_final == preds0, | |
| "staged_poison_rollbacks": staged_results, "pre_staging_rejections": pre_results, | |
| "honest_peer_review_of_poison": {"status": review.status, "reason": review.reason}, | |
| "legitimate_update_after_rollbacks": {"status": ok.status, "seq_after": ok.seq_after}, | |
| } | |