"""Comprehensive Test Suite for Armiger Modular CSAM-Safe Encrypted DHT DB Sync Plugin. Tests: 1. Multi-Layer CSAM & Harmful-Content Protection: - Multi-stage ContentFilter normalization parity (normalization, leetspeak, zero-width chars, diacritics) - Private tracker passkey stripping for complete anonymity - Direct CSAM & underage exploitation blocking - Non-consensual abuse & extreme violence blocking - Disguised malware dropper (.mp4.exe, .scr, .bat) blocking - Armiger Mondrian conformal safety gates (junk, out-of-scope, adult leakage) - Ed25519-signed blinded SHA-256 tombstone purge engine (verification, anti-replay, active purging) 2. Modular Sharded Subscription Architecture: - Partitioning by Category (10 modules) x Language (14 modules) x Quality Tier x Epoch - Strict isolation & opt-in for 'xxx' (default OFF, segregated table, no leakage) - Selective subscription filtering (only subscribed shards are processed) 3. X25519 + ChaCha20-Poly1305 AEAD Encryption: - HKDF key derivation - Round-trip encryption and decryption - AAD integrity protection (tampering, cross-shard transplant defense) 4. BEP-44 / BEP-46 Root-of-Trust + K-of-N Distinct-ASN Quorum + Merkle Shards: - BEP-44 DHT target derivation - Monotonic sequence verification - K-of-N quorum threshold & Sybil distinct-ASN enforcement - Merkle root verification and individual inclusion proofs 5. Anonymity & Per-Module Opt-In / Opt-Out: - Zero PII in canonical payloads - SyncMode (OFF, PULL_ONLY, SUBSCRIBE_AND_SEED, PUBLISHER) - Dynamic subscription modifications """ from __future__ import annotations import hashlib import os import sys import unittest from csam_guard import ( ArmigerSafetyGate, CSAMGuard, CSAMViolationType, MLKEM768, SignedTombstoneList, TombstoneStore, WOTSPlusSHAKE256, blind_infohash, generate_ed25519_keypair, normalize_text, sanitize_title, sign_tombstone_list, ) from armiger_dht_db_sync import ( ALL_CATEGORIES, CATEGORIES_MAINSTREAM, CATEGORY_ADULT, DEFAULT_EPOCH, LANGUAGES, QUALITY_TIERS, ArmigerDHTDatabaseSync, DHTShardManifest, MerkleTree, QuorumAttestation, QuorumVerifier, ShardCryptoEngine, SubscriptionConfig, SyncMode, TorrentRecord, create_mock_quorum_attestations, deserialize_shard_records, serialize_shard_records, ) from cryptography.exceptions import InvalidTag from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey class TestCSAMGuardAndSafetyGates(unittest.TestCase): """Test Suite 1: Multi-layer CSAM, harmful content, and safety gate protections.""" def setUp(self) -> None: self.root_priv, self.root_pub = generate_ed25519_keypair() self.tombstone_store = TombstoneStore(trusted_root_pubkeys_hex=[self.root_pub.hex()]) self.guard = CSAMGuard(tombstone_store=self.tombstone_store) self.safety_gate = ArmigerSafetyGate() def test_text_normalization_and_leetspeak(self) -> None: # Leetspeak deobfuscation raw = "P3d0ph1l1a" norm = normalize_text(raw) self.assertIn("pedophilia", norm) # Zero-width stripping and diacritics raw2 = "T\u200Be\u200De\u200Bn" self.assertEqual(normalize_text(raw2), "teen") # Repetitive runs raw3 = "teeeen porn" self.assertEqual(normalize_text(raw3), "teen porn") def test_private_tracker_passkey_sanitization(self) -> None: raw_title = "Big.Buck.Bunny.2008.1080p.BluRay[passkey=abc123secretkey]?passkey=xyz789" sanitized = sanitize_title(raw_title) self.assertNotIn("passkey=", sanitized) self.assertNotIn("abc123secretkey", sanitized) self.assertEqual(sanitized, "Big.Buck.Bunny.2008.1080p.BluRay") def test_direct_csam_detection(self) -> None: forbidden_samples = [ "Archive_Collection_CSAM_vids", "Underage porn leaked collection", "Pre-teen sex video archive", "Kindergarten erotic private camera", "Ch1ld p0rn direct link 2026", "Lolita hardcore uncensored stream", ] for sample in forbidden_samples: res = self.guard.inspect_torrent( infohash="a" * 40, title=sample, category="movie", ) self.assertFalse(res.is_safe, f"Failed to block CSAM sample: {sample}") self.assertEqual(res.violation_type, CSAMViolationType.CSAM_EXPLOITATION) def test_nonconsensual_and_extreme_violence_detection(self) -> None: # Non-consensual exploitation res1 = self.guard.inspect_torrent("b" * 40, "Secret Bathroom Spycam Upskirt Footage") self.assertFalse(res1.is_safe) self.assertEqual(res1.violation_type, CSAMViolationType.NONCONSENSUAL_ABUSE) # Extreme violence / snuff res2 = self.guard.inspect_torrent("c" * 40, "Real Snuff Film Execution Video 2026") self.assertFalse(res2.is_safe) self.assertEqual(res2.violation_type, CSAMViolationType.EXTREME_VIOLENCE_GORE) def test_disguised_malware_dropper_in_media_torrents(self) -> None: # Dangerous executable disguised as media in a movie torrent malicious_files = ["Big.Buck.Bunny.2008.1080p.mkv", "codec_update.mp4.exe", "instructions.txt"] res = self.guard.inspect_torrent( infohash="d" * 40, title="Big.Buck.Bunny.2008.1080p.BluRay.x264", file_list=malicious_files, category="movie", ) self.assertFalse(res.is_safe) self.assertEqual(res.violation_type, CSAMViolationType.MALWARE_PAYLOAD) self.assertIn(".exe", res.reason) def test_armiger_mondrian_safety_gate(self) -> None: # 1. High junk probability (malware/phishing) ok1, reason1, vtype1 = self.safety_gate.evaluate(p_adult=0.01, p_junk=0.15, p_excluded_scope=0.02, target_category="movie") self.assertFalse(ok1) self.assertEqual(vtype1, CSAMViolationType.SAFETY_GATE_JUNK) # 2. High out-of-scope probability ok2, reason2, vtype2 = self.safety_gate.evaluate(p_adult=0.01, p_junk=0.005, p_excluded_scope=0.55, target_category="movie") self.assertFalse(ok2) self.assertEqual(vtype2, CSAMViolationType.SAFETY_GATE_OUT_OF_SCOPE) # 3. Adult leakage into mainstream category ok3, reason3, vtype3 = self.safety_gate.evaluate(p_adult=0.88, p_junk=0.001, p_excluded_scope=0.01, target_category="movie") self.assertFalse(ok3) self.assertEqual(vtype3, CSAMViolationType.UNISOLATED_ADULT) # 4. Adult permitted when targeted at isolated 'xxx' category ok4, reason4, vtype4 = self.safety_gate.evaluate(p_adult=0.88, p_junk=0.001, p_excluded_scope=0.01, target_category="xxx") self.assertTrue(ok4) self.assertIsNone(vtype4) def test_signed_tombstone_purge_lifecycle(self) -> None: bad_hash_1 = "1111111111222222222233333333334444444444" bad_hash_2 = "aaaaaaaaaabbbbbbbbbbccccccccccdddddddddd" # Initially, hash is not tombstoned self.assertFalse(self.tombstone_store.is_tombstoned(bad_hash_1)) # Sign tombstone manifest with root key manifest = sign_tombstone_list( seq=1, infohashes_or_blinded=[bad_hash_1, bad_hash_2], private_key_bytes=self.root_priv, reasons={blind_infohash(bad_hash_1): "CSAM_ABUSE"}, ) ok, msg, count = self.tombstone_store.apply_signed_tombstones(manifest) self.assertTrue(ok) self.assertEqual(count, 2) self.assertTrue(self.tombstone_store.is_tombstoned(bad_hash_1)) self.assertTrue(self.tombstone_store.is_tombstoned(bad_hash_2)) # Anti-replay / downgrade check: seq=1 re-submitted must be rejected ok_dup, msg_dup, _ = self.tombstone_store.apply_signed_tombstones(manifest) self.assertFalse(ok_dup) self.assertIn("stale_or_replayed_tombstone_seq", msg_dup) # Guard immediately blocks ingestion of tombstoned hash res = self.guard.inspect_torrent(infohash=bad_hash_1, title="Harmless Looking Title") self.assertFalse(res.is_safe) self.assertEqual(res.violation_type, CSAMViolationType.TOMBSTONED_HASH) class TestModularShardedSubscription(unittest.TestCase): """Test Suite 2: Modular subscription partitioned by Category, Language, Quality, Epoch.""" def test_taxonomy_definitions(self) -> None: self.assertEqual(len(CATEGORIES_MAINSTREAM), 9) self.assertEqual(len(ALL_CATEGORIES), 10) self.assertEqual(CATEGORY_ADULT, "xxx") self.assertGreaterEqual(len(LANGUAGES), 14) self.assertGreaterEqual(len(QUALITY_TIERS), 7) def test_strict_xxx_isolation_in_subscription(self) -> None: # Default config: enable_xxx=False cfg = SubscriptionConfig( categories={"movie", "xxx"}, # Even if user includes "xxx" in category set languages={"en"}, quality_tiers={"hd_1080p"}, epochs={"2026_q4"}, enable_xxx=False, ) # Mainstream shard is subscribed self.assertTrue(cfg.is_subscribed("movie/en/hd_1080p/2026_q4")) # Adult shard is strictly blocked because enable_xxx is False self.assertFalse(cfg.is_subscribed("xxx/en/hd_1080p/2026_q4")) # Opt-in to adult: cfg.enable_xxx = True self.assertTrue(cfg.is_subscribed("xxx/en/hd_1080p/2026_q4")) def test_selective_shard_filtering(self) -> None: # Node subscribes only to English & Japanese movies in 1080p/4K cfg = SubscriptionConfig( categories={"movie"}, languages={"en", "ja"}, quality_tiers={"uhd_4k", "hd_1080p"}, epochs={"2026_q4"}, enable_xxx=False, ) # Matches self.assertTrue(cfg.is_subscribed("movie/en/hd_1080p/2026_q4")) self.assertTrue(cfg.is_subscribed("movie/ja/uhd_4k/2026_q4")) # Mismatches (ignored without bandwidth overhead) self.assertFalse(cfg.is_subscribed("movie/fr/hd_1080p/2026_q4")) # Wrong language self.assertFalse(cfg.is_subscribed("tv_show/en/hd_1080p/2026_q4")) # Wrong category self.assertFalse(cfg.is_subscribed("movie/en/sd/2026_q4")) # Wrong quality self.assertFalse(cfg.is_subscribed("movie/en/hd_1080p/2025_q1")) # Wrong epoch def test_database_table_segregation(self) -> None: sync_engine = ArmigerDHTDatabaseSync( subscription=SubscriptionConfig(enable_xxx=False), ) # Ingest mainstream record rec_main = TorrentRecord( infohash="1111222233334444555566667777888899990000", title="Tears.of.Steel.2012.1080p", category="movie", language="en", quality_tier="hd_1080p", epoch="2026_q4", size_bytes=10000000, file_count=1, seeders=150, leechers=10, content_sha256="abcd" * 16, file_list=["Tears.of.Steel.2012.1080p.mkv"], safety_gate_passed=True, created_at=100, ) self.assertTrue(sync_engine.db.insert_record(rec_main)) # Attempt to insert adult record into non-opted database rec_adult = TorrentRecord( infohash="9999888877776666555544443333222211110000", title="Adult.Feature.Film.XXX", category="xxx", language="en", quality_tier="hd_1080p", epoch="2026_q4", size_bytes=5000000, file_count=1, seeders=50, leechers=5, content_sha256="ef01" * 16, file_list=["Adult.Feature.Film.XXX.mp4"], safety_gate_passed=True, created_at=100, ) self.assertFalse(sync_engine.db.insert_record(rec_adult)) # Verify search only sees mainstream results = sync_engine.db.search("Tears.of.Steel") self.assertEqual(len(results), 1) self.assertEqual(results[0].category, "movie") # Adult search returns empty adult_results = sync_engine.db.search(category="xxx") self.assertEqual(len(adult_results), 0) sync_engine.close() class TestAEADEncryptionAndKeyDerivation(unittest.TestCase): """Test Suite 3: X25519 & ChaCha20-Poly1305 AEAD payload security.""" def setUp(self) -> None: self.master_secret = os.urandom(32) self.shard_id = "movie/en/hd_1080p/2026_q4" self.epoch = "2026_q4" self.seq = 42 self.merkle_root = "f" * 64 self.key = ShardCryptoEngine.derive_shard_key(self.master_secret, self.shard_id, self.epoch) self.sample_payload = b"Armiger Columnar SQLite Shard Binary Micro-Bundle" def test_key_derivation_deterministic_and_unique(self) -> None: # Same parameters -> same key key1 = ShardCryptoEngine.derive_shard_key(self.master_secret, self.shard_id, self.epoch) key2 = ShardCryptoEngine.derive_shard_key(self.master_secret, self.shard_id, self.epoch) self.assertEqual(key1, key2) self.assertEqual(len(key1), 32) # Different shard_id -> different key key_other = ShardCryptoEngine.derive_shard_key(self.master_secret, "tv_show/en/hd_1080p/2026_q4", self.epoch) self.assertNotEqual(key1, key_other) def test_round_trip_encryption_decryption(self) -> None: bundle = ShardCryptoEngine.encrypt_shard( plaintext=self.sample_payload, key=self.key, shard_id=self.shard_id, epoch=self.epoch, seq=self.seq, merkle_root=self.merkle_root, ) self.assertGreater(len(bundle), len(self.sample_payload)) decrypted = ShardCryptoEngine.decrypt_shard( bundle=bundle, key=self.key, shard_id=self.shard_id, epoch=self.epoch, seq=self.seq, merkle_root=self.merkle_root, ) self.assertEqual(decrypted, self.sample_payload) def test_ciphertext_tampering_defense(self) -> None: bundle = bytearray(ShardCryptoEngine.encrypt_shard( plaintext=self.sample_payload, key=self.key, shard_id=self.shard_id, epoch=self.epoch, seq=self.seq, merkle_root=self.merkle_root, )) # Tamper single byte in ciphertext bundle[-1] ^= 0x01 with self.assertRaises(InvalidTag): ShardCryptoEngine.decrypt_shard( bundle=bytes(bundle), key=self.key, shard_id=self.shard_id, epoch=self.epoch, seq=self.seq, merkle_root=self.merkle_root, ) def test_cross_shard_transplant_defense(self) -> None: # An attacker attempts to reuse valid ciphertext from 'movie' in 'game' shard bundle = ShardCryptoEngine.encrypt_shard( plaintext=self.sample_payload, key=self.key, shard_id=self.shard_id, epoch=self.epoch, seq=self.seq, merkle_root=self.merkle_root, ) with self.assertRaises(InvalidTag): # Attempting to decrypt with different shard_id in AAD fails ShardCryptoEngine.decrypt_shard( bundle=bundle, key=self.key, shard_id="game/en/hd_1080p/2026_q4", # Transplanted shard_id epoch=self.epoch, seq=self.seq, merkle_root=self.merkle_root, ) def test_post_quantum_mlkem768_hybrid_shard_encryption_and_fo_rejection(self) -> None: """Verify Hybrid X25519 + NIST FIPS 203 ML-KEM-768 shard encryption and FO implicit rejection.""" priv_pq, pub_pq = ShardCryptoEngine.generate_pq_keypair() self.assertEqual(len(priv_pq), 2432) # 32B X25519 + 2400B ML-KEM-768 dk self.assertEqual(len(pub_pq), 1216) # 32B X25519 + 1184B ML-KEM-768 ek pq_bundle = ShardCryptoEngine.encrypt_shard_pq( plaintext=self.sample_payload, recipient_hybrid_pubkey=pub_pq, shard_id=self.shard_id, epoch=self.epoch, seq=self.seq, merkle_root=self.merkle_root, ) decrypted = ShardCryptoEngine.decrypt_shard_pq( pq_bundle=pq_bundle, recipient_hybrid_privkey=priv_pq, shard_id=self.shard_id, epoch=self.epoch, seq=self.seq, merkle_root=self.merkle_root, ) self.assertEqual(decrypted, self.sample_payload) # Flip 1 bit inside the 1,088-byte ML-KEM-768 lattice ciphertext -> FO implicit rejection -> InvalidTag tampered = bytearray(pq_bundle) tampered[64] ^= 0x01 with self.assertRaises(InvalidTag): ShardCryptoEngine.decrypt_shard_pq( pq_bundle=bytes(tampered), recipient_hybrid_privkey=priv_pq, shard_id=self.shard_id, epoch=self.epoch, seq=self.seq, merkle_root=self.merkle_root, ) def test_post_quantum_wots_shake256_manifest_and_tombstone_tamper_rejection(self) -> None: """Verify WOTS+-SHAKE256 Post-Quantum dual-signatures block forged shard manifests and tombstones.""" import dataclasses root_priv, root_pub = generate_ed25519_keypair() tomb_store = TombstoneStore(trusted_root_pubkeys_hex=[root_pub.hex()]) tomb = sign_tombstone_list(seq=1, infohashes_or_blinded=["a" * 40], private_key_bytes=root_priv) self.assertEqual(len(bytes.fromhex(tomb.pq_pubkey_hex)), 32) self.assertEqual(len(bytes.fromhex(tomb.pq_signature_hex)), 2176) # Tamper 1 byte of the WOTS+-SHAKE256 PQ signature while keeping valid Ed25519 signature bad_pq_sig = bytearray(bytes.fromhex(tomb.pq_signature_hex)) bad_pq_sig[100] ^= 0xFF tampered_tomb = dataclasses.replace(tomb, pq_signature_hex=bad_pq_sig.hex()) ok, msg, _ = tomb_store.apply_signed_tombstones(tampered_tomb) self.assertFalse(ok) self.assertEqual(msg, "pq_wots_signature_verification_failed") class TestBEP44QuorumAndMerkleTree(unittest.TestCase): """Test Suite 4: BEP-44/46 Ed25519 root, K-of-N distinct-ASN quorum, and Merkle proofs.""" def setUp(self) -> None: self.root_priv, self.root_pub = generate_ed25519_keypair() self.verifier = QuorumVerifier( trusted_root_pubkeys_hex=[self.root_pub.hex()], min_quorum=5, min_distinct_asns=5, ) def test_merkle_tree_construction_and_inclusion_proofs(self) -> None: records = [ TorrentRecord( infohash=f"{i:040x}", title=f"Movie Title {i}", category="movie", language="en", quality_tier="hd_1080p", epoch="2026_q4", size_bytes=1000 * i, file_count=1, seeders=10 * i, leechers=i, content_sha256=f"{i:064x}", file_list=[f"movie_{i}.mkv"], safety_gate_passed=True, created_at=100 + i, ) for i in range(1, 8) # 7 records (tests odd-leaf handling) ] compressed, tree = serialize_shard_records(records) self.assertIsNotNone(tree.root_hex) # Verify inclusion proof for each record for idx in range(len(records)): leaf_hash = records[idx].leaf_hash() proof = tree.get_proof(idx) self.assertTrue(MerkleTree.verify_proof(leaf_hash, proof, tree.root_hex)) # Tampered leaf hash must fail proof verification fake_leaf = hashlib.sha256(b"tampered_record").digest() self.assertFalse(MerkleTree.verify_proof(fake_leaf, tree.get_proof(0), tree.root_hex)) def test_quorum_verification_success(self) -> None: # Create manifest manifest_dummy = DHTShardManifest( seq=1, shard_id="movie/en/hd_1080p/2026_q4", epoch="2026_q4", record_count=10, compressed_size_bytes=500, uncompressed_size_bytes=1200, merkle_root="a" * 64, ciphertext_sha256="b" * 64, bep46_infohash="c" * 40, timestamp=1000, root_pubkey_hex=self.root_pub.hex(), ed25519_signature_hex="", ) sk = Ed25519PrivateKey.from_private_bytes(self.root_priv) sig = sk.sign(manifest_dummy.canonical_payload()) # Generate 5 valid attestations from 5 distinct ASNs (Google, Cloudflare, Amazon, Comcast, Lumen) distinct_asns = [15169, 13335, 16509, 7922, 3356] attestations = create_mock_quorum_attestations(manifest_dummy.canonical_digest(), distinct_asns) manifest = DHTShardManifest( seq=manifest_dummy.seq, shard_id=manifest_dummy.shard_id, epoch=manifest_dummy.epoch, record_count=manifest_dummy.record_count, compressed_size_bytes=manifest_dummy.compressed_size_bytes, uncompressed_size_bytes=manifest_dummy.uncompressed_size_bytes, merkle_root=manifest_dummy.merkle_root, ciphertext_sha256=manifest_dummy.ciphertext_sha256, bep46_infohash=manifest_dummy.bep46_infohash, timestamp=manifest_dummy.timestamp, root_pubkey_hex=self.root_pub.hex(), ed25519_signature_hex=sig.hex(), attestations=attestations, ) ok, msg = self.verifier.verify_manifest(manifest, current_seq=0) self.assertTrue(ok) self.assertEqual(msg, "quorum_verified") def test_sybil_attack_rejected_duplicate_asns(self) -> None: manifest_dummy = DHTShardManifest( seq=1, shard_id="movie/en/hd_1080p/2026_q4", epoch="2026_q4", record_count=10, compressed_size_bytes=500, uncompressed_size_bytes=1200, merkle_root="a" * 64, ciphertext_sha256="b" * 64, bep46_infohash="c" * 40, timestamp=1000, root_pubkey_hex=self.root_pub.hex(), ed25519_signature_hex="", ) sk = Ed25519PrivateKey.from_private_bytes(self.root_priv) sig = sk.sign(manifest_dummy.canonical_payload()) # Sybil attacker creates 5 nodes but all hosted in the same ASN (e.g. AS99999) sybil_asns = [99999, 99999, 99999, 99999, 99999] attestations = create_mock_quorum_attestations(manifest_dummy.canonical_digest(), sybil_asns) manifest = DHTShardManifest( seq=manifest_dummy.seq, shard_id=manifest_dummy.shard_id, epoch=manifest_dummy.epoch, record_count=manifest_dummy.record_count, compressed_size_bytes=manifest_dummy.compressed_size_bytes, uncompressed_size_bytes=manifest_dummy.uncompressed_size_bytes, merkle_root=manifest_dummy.merkle_root, ciphertext_sha256=manifest_dummy.ciphertext_sha256, bep46_infohash=manifest_dummy.bep46_infohash, timestamp=manifest_dummy.timestamp, root_pubkey_hex=self.root_pub.hex(), ed25519_signature_hex=sig.hex(), attestations=attestations, ) ok, msg = self.verifier.verify_manifest(manifest, current_seq=0) self.assertFalse(ok) self.assertIn("insufficient_distinct_asns", msg) class TestEndToEndDHTSyncEngine(unittest.TestCase): """Test Suite 5: End-to-end sync, publisher-subscriber pipeline, purge, and modes.""" def setUp(self) -> None: self.root_priv, self.root_pub = generate_ed25519_keypair() self.master_secret = os.urandom(32) # Publisher instance self.publisher = ArmigerDHTDatabaseSync( subscription=SubscriptionConfig( categories={"movie", "tv_show", "anime"}, languages={"en", "ja"}, quality_tiers={"uhd_4k", "hd_1080p"}, epochs={"2026_q4"}, enable_xxx=False, sync_mode=SyncMode.PUBLISHER, ), master_secret=self.master_secret, trusted_root_pubkeys_hex=[self.root_pub.hex()], ) # Subscriber instance (Only subscribes to anime in ja) self.subscriber = ArmigerDHTDatabaseSync( subscription=SubscriptionConfig( categories={"anime"}, languages={"ja"}, quality_tiers={"hd_1080p"}, epochs={"2026_q4"}, enable_xxx=False, sync_mode=SyncMode.PULL_ONLY, min_quorum=5, min_distinct_asns=5, ), master_secret=self.master_secret, trusted_root_pubkeys_hex=[self.root_pub.hex()], ) def tearDown(self) -> None: if hasattr(self, "publisher"): self.publisher.close() if hasattr(self, "subscriber"): self.subscriber.close() def test_full_publish_subscribe_and_shard_application(self) -> None: # Ingest 3 anime records on publisher records_to_shard = [] for i in range(1, 4): ok, msg, rec = self.publisher.ingest_candidate_record( infohash=f"abcdef{i:034x}", title=f"Wing It Open Series S04E{i:02d} 1080p", category="anime", language="ja", quality_tier="hd_1080p", epoch="2026_q4", size_bytes=1500000000, file_count=1, seeders=200, leechers=15, p_adult=0.01, p_junk=0.001, p_excluded_scope=0.01, ) self.assertTrue(ok) self.assertIsNotNone(rec) records_to_shard.append(rec) # Build and sign shard on publisher shard_id = "anime/ja/hd_1080p/2026_q4" manifest, bundle = self.publisher.build_and_sign_shard( shard_id=shard_id, records=records_to_shard, root_privkey=self.root_priv, seq=1, ) # Add 5 distinct ASN attestations over the canonical manifest digest distinct_asns = [15169, 13335, 16509, 7922, 3356] attestations = create_mock_quorum_attestations(manifest.canonical_digest(), distinct_asns) manifest_with_quorum = manifest.with_attestations(attestations) # Subscriber applies shard ok, msg, count = self.subscriber.receive_and_apply_shard(manifest_with_quorum, bundle) self.assertTrue(ok, f"Subscriber failed to apply shard: {msg}") self.assertEqual(count, 3) # Verify records exist in subscriber database sub_records = self.subscriber.db.search("Wing It Open Series") self.assertEqual(len(sub_records), 3) # Test tombstone purge on subscriber tombstone_manifest = sign_tombstone_list( seq=1, infohashes_or_blinded=[records_to_shard[0].infohash], private_key_bytes=self.root_priv, ) ok_tomb, msg_tomb, new_tomb, purged = self.subscriber.apply_signed_tombstones(tombstone_manifest) self.assertTrue(ok_tomb) self.assertEqual(new_tomb, 1) self.assertEqual(purged, 1) # Database should now have only 2 records remaining = self.subscriber.db.search("Wing It Open Series") self.assertEqual(len(remaining), 2) self.assertNotIn(records_to_shard[0].infohash, [r.infohash for r in remaining]) def test_sync_mode_off_disables_all_operations(self) -> None: self.subscriber.set_sync_mode(SyncMode.OFF) # Candidate ingestion blocked ok1, msg1, _ = self.subscriber.ingest_candidate_record( infohash="1" * 40, title="Some Movie", category="movie", language="en", quality_tier="hd_1080p", ) self.assertFalse(ok1) self.assertEqual(msg1, "sync_mode_off") # Shard application blocked dummy_manifest = DHTShardManifest( seq=1, shard_id="anime/ja/hd_1080p/2026_q4", epoch="2026_q4", record_count=1, compressed_size_bytes=10, uncompressed_size_bytes=10, merkle_root="0" * 64, ciphertext_sha256="0" * 64, bep46_infohash="0" * 40, timestamp=100, root_pubkey_hex=self.root_pub.hex(), ed25519_signature_hex="0" * 128, ) ok2, msg2, count2 = self.subscriber.receive_and_apply_shard(dummy_manifest, b"") self.assertFalse(ok2) self.assertEqual(msg2, "sync_mode_off") def test_untrusted_root_key_rejected(self) -> None: untrusted_priv, untrusted_pub = generate_ed25519_keypair() manifest, bundle = self.publisher.build_and_sign_shard( shard_id="anime/ja/hd_1080p/2026_q4", records=[], root_privkey=untrusted_priv, seq=1, ) distinct_asns = [15169, 13335, 16509, 7922, 3356] attestations = create_mock_quorum_attestations(manifest.canonical_digest(), distinct_asns) manifest_with_quorum = manifest.with_attestations(attestations) ok, msg, count = self.subscriber.receive_and_apply_shard(manifest_with_quorum, bundle) self.assertFalse(ok) self.assertIn("untrusted_root_pubkey", msg) def test_stale_or_downgrade_sequence_rejected(self) -> None: shard_id = "anime/ja/hd_1080p/2026_q4" manifest1, bundle1 = self.publisher.build_and_sign_shard( shard_id=shard_id, records=[], root_privkey=self.root_priv, seq=5, ) distinct_asns = [15169, 13335, 16509, 7922, 3356] attestations1 = create_mock_quorum_attestations(manifest1.canonical_digest(), distinct_asns) manifest1 = manifest1.with_attestations(attestations1) ok1, _, _ = self.subscriber.receive_and_apply_shard(manifest1, bundle1) self.assertTrue(ok1) # Re-send seq=5 (replay) or seq=4 (downgrade) manifest_stale, bundle_stale = self.publisher.build_and_sign_shard( shard_id=shard_id, records=[], root_privkey=self.root_priv, seq=5, ) attestations_stale = create_mock_quorum_attestations(manifest_stale.canonical_digest(), distinct_asns) manifest_stale = manifest_stale.with_attestations(attestations_stale) ok2, msg2, _ = self.subscriber.receive_and_apply_shard(manifest_stale, bundle_stale) self.assertFalse(ok2) self.assertIn("stale_or_downgrade_sequence", msg2) def test_ciphertext_sha256_mismatch_rejected(self) -> None: shard_id = "anime/ja/hd_1080p/2026_q4" manifest, bundle = self.publisher.build_and_sign_shard( shard_id=shard_id, records=[], root_privkey=self.root_priv, seq=10, ) distinct_asns = [15169, 13335, 16509, 7922, 3356] manifest = manifest.with_attestations(create_mock_quorum_attestations(manifest.canonical_digest(), distinct_asns)) # Corrupted bundle bytes in transit corrupted_bundle = bundle[:-1] + bytes([bundle[-1] ^ 0xFF]) ok, msg, _ = self.subscriber.receive_and_apply_shard(manifest, corrupted_bundle) self.assertFalse(ok) self.assertEqual(msg, "ciphertext_sha256_mismatch") def test_merkle_root_mismatch_rejected(self) -> None: shard_id = "anime/ja/hd_1080p/2026_q4" # Generate shard with 1 record rec = TorrentRecord( infohash="1" * 40, title="Anime Episode 1", category="anime", language="ja", quality_tier="hd_1080p", epoch="2026_q4", size_bytes=100, file_count=1, seeders=10, leechers=1, content_sha256="2" * 64, file_list=["ep1.mkv"], safety_gate_passed=True, created_at=100, ) manifest, bundle = self.publisher.build_and_sign_shard( shard_id=shard_id, records=[rec], root_privkey=self.root_priv, seq=20, ) # Forge manifest with incorrect merkle_root import dataclasses tampered_manifest = dataclasses.replace(manifest, merkle_root="9" * 64) sk = Ed25519PrivateKey.from_private_bytes(self.root_priv) new_sig = sk.sign(tampered_manifest.canonical_payload()) pq_priv, _ = WOTSPlusSHAKE256.derive_keypair_from_ed25519_priv(self.root_priv) new_pq_sig = WOTSPlusSHAKE256.sign(pq_priv, tampered_manifest.canonical_payload()) tampered_manifest = dataclasses.replace( tampered_manifest, ed25519_signature_hex=new_sig.hex(), pq_signature_hex=new_pq_sig.hex(), ) distinct_asns = [15169, 13335, 16509, 7922, 3356] tampered_manifest = tampered_manifest.with_attestations( create_mock_quorum_attestations(tampered_manifest.canonical_digest(), distinct_asns) ) ok, msg, _ = self.subscriber.receive_and_apply_shard(tampered_manifest, bundle) self.assertFalse(ok) self.assertIn("aead_decryption_failed", msg) # Because AAD checks merkle_root! def test_pre_ingest_reinspection_drops_bad_records_in_shard(self) -> None: shard_id = "anime/ja/hd_1080p/2026_q4" # A shard contains 1 valid record and 1 disguised malware dropper rec_clean = TorrentRecord( infohash="c1ea0" + "0" * 35, title="Clean Anime Movie", category="anime", language="ja", quality_tier="hd_1080p", epoch="2026_q4", size_bytes=100, file_count=1, seeders=10, leechers=1, content_sha256="1" * 64, file_list=["movie.mkv"], safety_gate_passed=True, created_at=100, ) rec_malware = TorrentRecord( infohash="bad0000" + "0" * 33, title="Clean Anime Movie Part 2", category="anime", language="ja", quality_tier="hd_1080p", epoch="2026_q4", size_bytes=100, file_count=1, seeders=10, leechers=1, content_sha256="2" * 64, file_list=["movie.mkv.exe"], # Disguised malware dropper safety_gate_passed=True, created_at=100, ) manifest, bundle = self.publisher.build_and_sign_shard( shard_id=shard_id, records=[rec_clean, rec_malware], root_privkey=self.root_priv, seq=30, ) distinct_asns = [15169, 13335, 16509, 7922, 3356] manifest = manifest.with_attestations( create_mock_quorum_attestations(manifest.canonical_digest(), distinct_asns) ) ok, msg, count = self.subscriber.receive_and_apply_shard(manifest, bundle) self.assertTrue(ok) # Only clean record is ingested, malware dropped! self.assertEqual(count, 1) self.assertIsNotNone(self.subscriber.db.get_record(rec_clean.infohash)) self.assertIsNone(self.subscriber.db.get_record(rec_malware.infohash)) def test_x25519_ecdh_key_exchange(self) -> None: from cryptography.hazmat.primitives.asymmetric.x25519 import X25519PrivateKey alice_priv = X25519PrivateKey.generate() bob_priv = X25519PrivateKey.generate() alice_shared = alice_priv.exchange(bob_priv.public_key()) bob_shared = bob_priv.exchange(alice_priv.public_key()) self.assertEqual(alice_shared, bob_shared) key_alice = ShardCryptoEngine.derive_shard_key(alice_shared, "movie/en/hd_1080p/2026_q4", "2026_q4") key_bob = ShardCryptoEngine.derive_shard_key(bob_shared, "movie/en/hd_1080p/2026_q4", "2026_q4") self.assertEqual(key_alice, key_bob) def test_bep44_target_derivation(self) -> None: manifest_dummy = DHTShardManifest( seq=1, shard_id="movie/en/hd_1080p/2026_q4", epoch="2026_q4", record_count=0, compressed_size_bytes=0, uncompressed_size_bytes=0, merkle_root="0" * 64, ciphertext_sha256="0" * 64, bep46_infohash="0" * 40, timestamp=100, root_pubkey_hex=self.root_pub.hex(), ed25519_signature_hex="0" * 128, ) target = manifest_dummy.bep44_target() self.assertEqual(len(target), 40) # Deterministic self.assertEqual(target, manifest_dummy.bep44_target()) class TestPhase2CatalogRedTeamAndBenchmarks(unittest.TestCase): """Test Suite 6 (Phase 2): 10,000-Item Catalog Sharding, 120+ Adversarial Red-Team Traps & Benchmarks.""" def setUp(self) -> None: self.root_priv, self.root_pub = generate_ed25519_keypair() self.master_secret = os.urandom(32) self.tombstone_store = TombstoneStore(trusted_root_pubkeys_hex=[self.root_pub.hex()]) self.guard = CSAMGuard(tombstone_store=self.tombstone_store) self.safety_gate = ArmigerSafetyGate() def test_10000_item_catalog_sharding_selective_sync_and_merkle_proofs(self) -> None: """Verify 10,000-item multi-category/multi-language sharding, selective sync, 0% xxx leakage, and 10k Merkle proofs.""" categories = list(ALL_CATEGORIES) # 10 categories (9 mainstream + 'xxx') languages = ["en", "ja", "es", "fr", "de"] # 5 languages qualities = ["uhd_4k", "hd_1080p"] # 2 quality tiers epoch = "2026_q4" publisher = ArmigerDHTDatabaseSync( subscription=SubscriptionConfig( categories=set(categories), languages=set(languages), quality_tiers=set(qualities), epochs={epoch}, enable_xxx=True, sync_mode=SyncMode.PUBLISHER, ), master_secret=self.master_secret, trusted_root_pubkeys_hex=[self.root_pub.hex()], ) subscriber = ArmigerDHTDatabaseSync( subscription=SubscriptionConfig( categories={"movie", "tv_show", "anime"}, languages={"en", "ja"}, quality_tiers={"uhd_4k", "hd_1080p"}, epochs={epoch}, enable_xxx=False, sync_mode=SyncMode.PULL_ONLY, min_quorum=5, min_distinct_asns=5, ), master_secret=self.master_secret, trusted_root_pubkeys_hex=[self.root_pub.hex()], ) try: # 1. Generate 10,000 catalog items partitioned into 100 shards (100 items per shard) shards_map: dict[str, list[TorrentRecord]] = {} total_generated = 0 for c_idx, cat in enumerate(categories): for l_idx, lang in enumerate(languages): for q_idx, qual in enumerate(qualities): shard_id = SubscriptionConfig.format_shard_id(cat, lang, qual, epoch) shard_records: list[TorrentRecord] = [] for item_idx in range(100): global_id = (c_idx * 1000) + (l_idx * 200) + (q_idx * 100) + item_idx + 1 infohash = f"{global_id:040x}" rec = TorrentRecord( infohash=infohash, title=f"Catalog_{cat}_{lang}_{qual}_Item_{global_id:05d}", category=cat, language=lang, quality_tier=qual, epoch=epoch, size_bytes=1_000_000 * (item_idx + 1), file_count=1, seeders=50 + item_idx, leechers=5, content_sha256=f"{global_id:064x}", file_list=[f"item_{global_id:05d}.mkv"], safety_gate_passed=True, created_at=1700000000 + global_id, ) shard_records.append(rec) total_generated += 1 shards_map[shard_id] = shard_records self.assertEqual(total_generated, 10_000) self.assertEqual(len(shards_map), 100) # 2. Build & sign all 100 shards, verify 10,000 Merkle inclusion proofs, and feed to subscriber distinct_asns = [15169, 13335, 16509, 7922, 3356] reusable_node_keys = [generate_ed25519_keypair() for _ in distinct_asns] merkle_proofs_verified = 0 subscribed_shards_applied = 0 unsubscribed_shards_skipped = 0 records_ingested = 0 for shard_id, recs in shards_map.items(): # Verify binary Merkle tree inclusion proof for every record in the shard sorted_recs = sorted(recs, key=lambda r: r.infohash.lower()) leaves = [r.leaf_hash() for r in sorted_recs] tree = MerkleTree(leaves) for idx, r in enumerate(sorted_recs): proof = tree.get_proof(idx) self.assertTrue(MerkleTree.verify_proof(r.leaf_hash(), proof, tree.root_hex)) merkle_proofs_verified += 1 manifest, bundle = publisher.build_and_sign_shard( shard_id=shard_id, records=recs, root_privkey=self.root_priv, seq=1, ) digest = manifest.canonical_digest() attestations = [] for asn, (n_priv, n_pub) in zip(distinct_asns, reusable_node_keys): sk = Ed25519PrivateKey.from_private_bytes(n_priv) sig = sk.sign(digest) attestations.append( QuorumAttestation( node_id=hashlib.sha1(n_pub + f"asn{asn}".encode()).hexdigest(), asn=asn, ip_digest=hashlib.sha256(f"198.51.100.{asn % 250}".encode()).hexdigest(), pubkey_hex=n_pub.hex(), signature_hex=sig.hex(), timestamp=manifest.timestamp, ) ) manifest_quorum = manifest.with_attestations(attestations) ok, msg, count = subscriber.receive_and_apply_shard(manifest_quorum, bundle) cat, lang, qual, _ = SubscriptionConfig.parse_shard_id(shard_id) expected_match = (cat in {"movie", "tv_show", "anime"}) and (lang in {"en", "ja"}) if expected_match: self.assertTrue(ok, f"Expected shard {shard_id} to be applied, got: {msg}") self.assertEqual(count, 100) subscribed_shards_applied += 1 records_ingested += count else: self.assertFalse(ok) self.assertIn("unsubscribed_shard_filtered", msg) self.assertEqual(count, 0) unsubscribed_shards_skipped += 1 self.assertEqual(merkle_proofs_verified, 10_000) self.assertEqual(subscribed_shards_applied, 12) # 3 categories * 2 languages * 2 qualities self.assertEqual(unsubscribed_shards_skipped, 88) self.assertEqual(records_ingested, 1_200) # 3. Verify SQLite state & 0% 'xxx' leakage self.assertEqual(subscriber.db.count_records(), 1_200) self.assertEqual(subscriber.db.count_records(category="movie"), 400) self.assertEqual(subscriber.db.count_records(category="tv_show"), 400) self.assertEqual(subscriber.db.count_records(category="anime"), 400) self.assertEqual(subscriber.db.count_records(category="xxx"), 0) cur = subscriber.db.conn.cursor() cur.execute("SELECT COUNT(*) FROM mainstream_torrents WHERE category = 'xxx'") self.assertEqual(cur.fetchone()[0], 0) cur.execute("SELECT DISTINCT category, language FROM mainstream_torrents") pairs = {(r[0], r[1]) for r in cur.fetchall()} self.assertEqual( pairs, { ("movie", "en"), ("movie", "ja"), ("tv_show", "en"), ("tv_show", "ja"), ("anime", "en"), ("anime", "ja"), }, ) finally: publisher.close() subscriber.close() def test_adversarial_csam_nonconsensual_malware_redteam_100_plus_traps(self) -> None: """Red-team 120+ obfuscated CSAM, non-consensual, extreme violence, malware, conformal, and tombstone traps.""" total_traps_verified = 0 # Surface A: 25 Unicode NFKD, Fullwidth, Mathematical & Cyrillic/Greek Homoglyph Traps homoglyph_traps = [ "\uff23\uff33\uff21\uff2d Archive 2026", # Fullwidth CSAM "\uff23\uff33\uff21\uff25 Collection", # Fullwidth CSAE "\uff30\uff45\uff44\uff4f\uff50\uff48\uff49\uff4c\uff45", # Fullwidth Pedophile "\U0001d429\U0001d41e\U0001d41d\U0001d428\U0001d429\U0001d421\U0001d422\U0001d425\U0001d41e", # Math Bold "\U0001d450\U0001d455\U0001d456\U0001d459\U0001d451 \U0001d45d\U0001d45c\U0001d45f\U0001d45b", # Math Italic child porn "\U0001d52d\U0001d52f\U0001d522\U0001d531\U0001d522\U0001d522\U0001d52b \U0001d530\U0001d522\U0001d535", # Math Fraktur preteen sex "p\u00e9d\u00f6ph\u00efl\u00eb hidden pack", # Diacritic storm "pr\u00ebt\u00e9\u00e8n s\u00eax leak", # Diacritics preteen sex "l\u00f6l\u00eft\u00e4 h\u00e4rdc\u00f6r\u00eb", # Diacritics lolita hardcore "\u0440\u0435d\u043e\u0440hil\u0435 underground", # Cyrillic р, е, о "\u0441hild \u0440\u043ern video", # Cyrillic с, р, о "l\u043elit\u0430 h\u0430rd\u0441\u043er\u0435", # Cyrillic о, а, с, е "\u0455h\u043et\u0430\u0441\u043en collection", # Cyrillic ѕ, о, а, с "t\u043eddler \u0440\u043ern archive", # Cyrillic о, р "kinderg\u0430rten \u0455\u0435\u0445", # Cyrillic а, ѕ, е, х "inf\u0430nt \u0435r\u043eti\u0441", # Cyrillic а, е, о, с "und\u0435r\u0430ge \u0440\u043ern", # Cyrillic е, а, р, о "j\u0430ilb\u0430it pack", # Cyrillic а "in\u0441\u0435\u0455t \u0441hild", # Cyrillic с, е, ѕ "r\u0430\u0440\u0435 vid\u0435\u043e", # Cyrillic а, р, е, о "r\u0435v\u0435ng\u0435 \u0440\u043ern", # Cyrillic е, р, о "\u0455\u0440\u0443\u0441\u0430m u\u0440\u0455kirt", # Cyrillic ѕ, р, у, с, а "\u0455nuff film uncensored", # Cyrillic ѕ "r\u0435\u0430l b\u0435h\u0435\u0430ding", # Cyrillic е, а "\u03c1\u03b5d\u03bf\u03c1hil\u03b5 archive", # Greek ρ, ε, ο ] self.assertEqual(len(homoglyph_traps), 25) for sample in homoglyph_traps: res = self.guard.inspect_torrent("a" * 40, sample, category="movie") self.assertFalse(res.is_safe, f"Homoglyph trap bypassed guard: {sample!r}") total_traps_verified += 1 # Surface B: 25 Zero-Width Joiner, Soft Hyphen & Bidirectional Override Traps zw_bidi_traps = [ "c\u200bs\u200ca\u200dm leak", "c\u2060s\u2066a\u2069e dump", "p\u200be\u200cd\u200do\u200ep\u200fh\u202ai\u202el\u202ce", "ch\u200bild p\u200dorn", "cp\u200b v\u200cids", "cp\u2060 l\u2060inks", "cp\ufeff m\ufeffegas", "ch\u00adild er\u00adotic", "under\u200bage p\u200dorn", "pre\u200bteen s\u200dex", "pre\u202eteen n\u202cude", "loli\u200bta hard\u200ccore", "loli\u200dta hen\u200btai", "sho\u200btacon archive", "todd\u200bler p\u200dorn", "kinder\u200bgarten s\u200cex", "in\u200bfant n\u200dude", "elemen\u200btary er\u200cotic", "jail\u200bbait collection", "in\u200bcest mi\u200dnor", "non\u200bconsensual leak", "rape\u200b v\u200dideo", "revenge\u200b p\u200corn", "snuff\u200b f\u200dilm", "real\u200b ex\u202eecution", ] self.assertEqual(len(zw_bidi_traps), 25) for sample in zw_bidi_traps: res = self.guard.inspect_torrent("b" * 40, sample, category="movie") self.assertFalse(res.is_safe, f"Zero-width/Bidi trap bypassed guard: {sample!r}") total_traps_verified += 1 # Surface C: 25 Heavy Leetspeak, Character Repetition & Delimiter Spacing Traps leet_traps = [ "P3d0ph1l14_v1ds", "c_h_1_l_d__p_0_r_n", "pr3333t333n s333x", "l0l1t@ h@rdc0r3", "j@1lb@1t_2026", "r3v3ng3 p0rn", "5nuff f1lm", "b1@ckm@1l s3x", "5pyc@m uP5k1rt", "g@ng r@p3", "f0rc3d s3x", "c.s.a.m.pack", "c-s-a-e-collection", "k1nd3rg@rt3n p0rn", "t0ddl3r s3x", "1nf@nt p0rn", "3l3m3nt@ry 3r0t1c", "53xt0rt10n_leaks", "r3@l b3h3@d1ng", "r3@l 3x3cut10n", "l1v3 5u1c1d3 v1d30", "m@55@cr3 l1v35tr3@m", "r3@l t0rtur3 f1lm", "cru5h f3t15h k1ll", "1nc35t ch1ld", ] self.assertEqual(len(leet_traps), 25) for sample in leet_traps: res = self.guard.inspect_torrent("c" * 40, sample, category="movie") self.assertFalse(res.is_safe, f"Leetspeak trap bypassed guard: {sample!r}") total_traps_verified += 1 # Surface D: 25 Nested Archive & Double-Extension Disguised Executable Malware Traps malware_file_traps = [ ("movie", ["Movie.2026.1080p.mp4.exe"]), ("tv_show", ["Series.S01E01.mkv.scr"]), ("anime", ["Episode_01.avi.bat"]), ("book", ["Complete_Guide.pdf.vbs"]), ("music", ["01_Lossless_Track.flac.ps1"]), ("music", ["Album_Track_02.mp3.cmd"]), ("book", ["Novel_Edition.epub.hta"]), ("comic", ["Issue_001.cbz.jar"]), ("movie", ["Trailer_4K.mov.msi"]), ("comic", ["Cover_Art.jpg.pif"]), ("tv_show", ["Subtitles_EN.srt.wsf"]), ("music", ["Studio_Master.wav.vbe"]), ("music", ["Podcast_Ep1.m4a.jse"]), ("movie", ["WebRip_Stream.webm.com"]), ("tv_show", ["Remux_2160p.mkv.cpl"]), ("anime", ["OVA_Special.avi.wsh"]), ("movie", ["Feature_Film.mp4.psm1"]), ("book", ["Reference_Manual.pdf.application"]), ("music", ["Bonus_Track.flac.gadget"]), ("tv_show", ["Finale_1080p.mkv.msp"]), ("Archive.zip/Nested/Video.mp4.exe", ["Archive.zip/Nested/Video.mp4.exe"]), ("Season1.rar/S01E02.mkv.scr", ["Season1.rar/S01E02.mkv.scr"]), ("Pack.7z/Clip.avi.bat", ["Pack.7z/Clip.avi.bat"]), ("Library.tar.gz/Book.pdf.vbs", ["Library.tar.gz/Book.pdf.vbs"]), ("Disc.iso/Audio/01.flac.ps1", ["Disc.iso/Audio/01.flac.ps1"]), ] self.assertEqual(len(malware_file_traps), 25) for cat_or_path, flist in malware_file_traps: cat = cat_or_path if cat_or_path in ALL_CATEGORIES else None res = self.guard.inspect_torrent( infohash="d" * 40, title="Clean_Release_Title_2026", file_list=flist, category=cat, ) self.assertFalse(res.is_safe, f"Malware dropper bypassed guard: {flist}") self.assertEqual(res.violation_type, CSAMViolationType.MALWARE_PAYLOAD) total_traps_verified += 1 # Surface E: 20 Mondrian Conformal Safety Gate Boundary Traps # E1: Safe just below thresholds ok_safe, _, _ = self.safety_gate.evaluate(p_adult=0.5450, p_junk=0.1009, p_excluded_scope=0.4769, target_category="movie") self.assertTrue(ok_safe) conformal_traps = [ # 5 junk boundary & high-risk cases (threshold = 0.1010) (0.01, 0.1010, 0.01, "movie", CSAMViolationType.SAFETY_GATE_JUNK), (0.01, 0.1011, 0.01, "tv_show", CSAMViolationType.SAFETY_GATE_JUNK), (0.01, 0.1500, 0.01, "anime", CSAMViolationType.SAFETY_GATE_JUNK), (0.01, 0.5000, 0.01, "software", CSAMViolationType.SAFETY_GATE_JUNK), (0.01, 0.9900, 0.01, "game", CSAMViolationType.SAFETY_GATE_JUNK), # 6 scope boundary & high-risk cases (threshold = 0.4770) (0.01, 0.01, 0.4770, "movie", CSAMViolationType.SAFETY_GATE_OUT_OF_SCOPE), (0.01, 0.01, 0.4771, "tv_show", CSAMViolationType.SAFETY_GATE_OUT_OF_SCOPE), (0.01, 0.01, 0.5500, "music", CSAMViolationType.SAFETY_GATE_OUT_OF_SCOPE), (0.01, 0.01, 0.7000, "book", CSAMViolationType.SAFETY_GATE_OUT_OF_SCOPE), (0.01, 0.01, 0.8500, "comic", CSAMViolationType.SAFETY_GATE_OUT_OF_SCOPE), (0.01, 0.01, 0.9900, "course", CSAMViolationType.SAFETY_GATE_OUT_OF_SCOPE), # 9 unisolated adult boundary cases across all 9 mainstream categories (threshold = 0.5451) (0.5451, 0.01, 0.01, "movie", CSAMViolationType.UNISOLATED_ADULT), (0.5451, 0.01, 0.01, "tv_show", CSAMViolationType.UNISOLATED_ADULT), (0.5451, 0.01, 0.01, "music", CSAMViolationType.UNISOLATED_ADULT), (0.5451, 0.01, 0.01, "game", CSAMViolationType.UNISOLATED_ADULT), (0.5451, 0.01, 0.01, "software", CSAMViolationType.UNISOLATED_ADULT), (0.5451, 0.01, 0.01, "anime", CSAMViolationType.UNISOLATED_ADULT), (0.5451, 0.01, 0.01, "book", CSAMViolationType.UNISOLATED_ADULT), (0.5451, 0.01, 0.01, "comic", CSAMViolationType.UNISOLATED_ADULT), (0.5451, 0.01, 0.01, "course", CSAMViolationType.UNISOLATED_ADULT), ] self.assertEqual(len(conformal_traps), 20) for p_a, p_j, p_s, cat, expected_vtype in conformal_traps: ok, reason, vtype = self.safety_gate.evaluate(p_a, p_j, p_s, cat) self.assertFalse(ok, f"Conformal trap bypassed gate: {(p_a, p_j, p_s, cat)}") self.assertEqual(vtype, expected_vtype) total_traps_verified += 1 # Verify total adversarial traps >= 120 self.assertGreaterEqual(total_traps_verified, 120) # Surface F: Live Ed25519-Signed Blinded SHA-256 Tombstone Purge Propagation & Permanent Block engine = ArmigerDHTDatabaseSync( subscription=SubscriptionConfig( categories={"movie"}, languages={"en"}, quality_tiers={"hd_1080p"}, epochs={"2026_q4"}, enable_xxx=False, sync_mode=SyncMode.PUBLISHER, ), master_secret=self.master_secret, trusted_root_pubkeys_hex=[self.root_pub.hex()], ) try: all_hashes = [f"{i:040x}" for i in range(500, 550)] for idx, h in enumerate(all_hashes): ok, _, _ = engine.ingest_candidate_record( infohash=h, title=f"Verified_Mainstream_Film_{idx}", category="movie", language="en", quality_tier="hd_1080p", epoch="2026_q4", ) self.assertTrue(ok) self.assertEqual(engine.db.count_records(), 50) revoked_hashes = all_hashes[:15] tomb_manifest = sign_tombstone_list( seq=10, infohashes_or_blinded=revoked_hashes, private_key_bytes=self.root_priv, ) ok_t, msg_t, new_t, purged_t = engine.apply_signed_tombstones(tomb_manifest) self.assertTrue(ok_t) self.assertEqual(new_t, 15) self.assertEqual(purged_t, 15) self.assertEqual(engine.db.count_records(), 35) # Confirm permanent re-ingest block on all 15 tombstoned infohashes for rh in revoked_hashes: ok_re, msg_re, _ = engine.ingest_candidate_record( infohash=rh, title="Innocent_Renamed_Film", category="movie", language="en", quality_tier="hd_1080p", epoch="2026_q4", ) self.assertFalse(ok_re) self.assertIn("tombstoned_hash", msg_re) finally: engine.close() def test_throughput_and_latency_benchmarks(self) -> None: """Measure CSAM Guard items/sec, AEAD encryption/decryption MB/s, and SQLite batch ingest speed.""" import time # 1. CSAM Guard inspection throughput (5,000 items) n_inspect = 5_000 sample_titles = [ ("Clean.Movie.2026.2160p.WEB-DL.DDP5.1.Atmos.H.265", ["Movie.2026.2160p.mkv", "subs.srt"], True), ("Ch1ld_p0rn_hidden_archive_2026", ["archive.zip"], False), ("Disguised.Release.1080p", ["Release.1080p.mp4.exe"], False), ("Anime.Series.S02E12.1080p.HEVC", ["Anime.S02E12.mkv"], True), ("Secret_Bathroom_Spycam_Upskirt", ["clip.mp4"], False), ] t0 = time.perf_counter() for i in range(n_inspect): title, flist, expected_safe = sample_titles[i % len(sample_titles)] res = self.guard.inspect_torrent(f"{i + 10000:040x}", title, file_list=flist, category="movie") self.assertEqual(res.is_safe, expected_safe) t_inspect = time.perf_counter() - t0 inspect_rate = n_inspect / max(t_inspect, 1e-9) # 2. AEAD Encryption & Decryption Throughput (4 MB payload x 5 iterations = 20 MB) shard_id = "movie/en/uhd_4k/2026_q4" epoch = "2026_q4" merkle_root = "a" * 64 key = ShardCryptoEngine.derive_shard_key(self.master_secret, shard_id, epoch) payload_4mb = os.urandom(4 * 1024 * 1024) iters = 5 total_mb = (len(payload_4mb) * iters) / (1024 * 1024) t_enc0 = time.perf_counter() bundles = [ ShardCryptoEngine.encrypt_shard(payload_4mb, key, shard_id, epoch, seq=i + 1, merkle_root=merkle_root) for i in range(iters) ] t_enc = time.perf_counter() - t_enc0 enc_mbps = total_mb / max(t_enc, 1e-9) t_dec0 = time.perf_counter() for i, b in enumerate(bundles): pt = ShardCryptoEngine.decrypt_shard(b, key, shard_id, epoch, seq=i + 1, merkle_root=merkle_root) self.assertEqual(len(pt), len(payload_4mb)) t_dec = time.perf_counter() - t_dec0 dec_mbps = total_mb / max(t_dec, 1e-9) # 3. SQLite Batch Ingest Throughput (10,000 records) n_sqlite = 10_000 batch_records = [ TorrentRecord( infohash=f"{i + 100000:040x}", title=f"Benchmark_Record_{i:05d}", category="movie", language="en", quality_tier="hd_1080p", epoch="2026_q4", size_bytes=2_000_000_000, file_count=2, seeders=100, leechers=10, content_sha256=f"{i + 100000:064x}", file_list=[f"movie_{i:05d}.mkv", f"movie_{i:05d}.srt"], safety_gate_passed=True, created_at=1700000000 + i, ) for i in range(n_sqlite) ] engine = ArmigerDHTDatabaseSync( subscription=SubscriptionConfig(enable_xxx=False), master_secret=self.master_secret, trusted_root_pubkeys_hex=[self.root_pub.hex()], ) try: t_db0 = time.perf_counter() inserted = engine.db.insert_records_batch(batch_records) t_db = time.perf_counter() - t_db0 sqlite_rate = inserted / max(t_db, 1e-9) self.assertEqual(inserted, n_sqlite) self.assertEqual(engine.db.count_records(), n_sqlite) finally: engine.close() # Assert conservative floors to catch performance regressions across hosts self.assertGreater(inspect_rate, 5_000.0) self.assertGreater(enc_mbps, 50.0) self.assertGreater(dec_mbps, 50.0) self.assertGreater(sqlite_rate, 20_000.0) print( f"\n[BENCHMARK] CSAMGuard={inspect_rate:,.0f} items/s ({1e6/inspect_rate:.1f} us/item) | " f"AEAD_Enc={enc_mbps:,.1f} MB/s | AEAD_Dec={dec_mbps:,.1f} MB/s | " f"SQLite_Batch={sqlite_rate:,.0f} records/s" ) if __name__ == "__main__": unittest.main()