test_dht_db_sync.py

"""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()