armiger_dht_db_sync.py

"""Modular, CSAM-Protected, Encrypted, Poison/Spoof-Proof BEP-44 / BEP-46 DHT DB Sync Plugin.

Enforces:
  1. Multi-Layer CSAM & Abuse Protection:
     - Multi-stage ContentFilter normalization parity (normalization, leetspeak, zero-width stripping)
     - Deterministic & structural pre-ingest CSAM/abuse blocker
     - Ed25519-signed blinded SHA-256 tombstone purge engine
     - Armiger Mondrian conformal safety gates (junk/malware, out-of-scope, adult containment)
  2. Modular Sharded Subscription Architecture:
     - Partitioned by Category (10 modules, xxx strictly isolated & opt-in) x
       Language (14 modules) x Quality Tier (7 tiers) x Time Epoch
     - Fine-grained subscription filters: nodes only fetch and store subscribed shards
  3. X25519 + ChaCha20-Poly1305 AEAD Encryption:
     - Ephemeral / topic-derived key exchange via HKDF-SHA256
     - 12-byte random nonces and AAD binding shard ID, epoch, sequence, and Merkle root
     - Tamper-proof ciphertext with replay and cross-shard transplant defense
  4. BEP-44 / BEP-46 Root-of-Trust + K-of-N Distinct-ASN Quorum:
     - Strictly monotonic sequence counters (anti-replay/anti-downgrade)
     - Offline Ed25519 root signatures over canonical BEP-44 mutable manifests
     - Sybil-resistant K-of-N attestation quorum requiring independent distinct ASNs
     - Binary Merkle tree root verification and individual record inclusion proofs
  5. Anonymity & Per-Module Opt-In/Opt-Out:
     - Zero PII, peer IPs, or tracker credentials in payloads
     - Blinded infohash lookups and sanitized titles
     - Full sync modes: OFF, PULL_ONLY, SUBSCRIBE_AND_SEED, PUBLISHER
"""
from __future__ import annotations

import dataclasses
import enum
import hashlib
import hmac
import json
import os
import pathlib
import sqlite3
import time
import zlib
from typing import Any, Iterable, Mapping, Sequence
from armiger_dht_rsi import (
    AnonymityGuard,
    AnonymityViolationError,
    _bep44_canonical_sign_buffer,
    resolve_opt_in_sharing,
)

from cryptography.exceptions import InvalidSignature, InvalidTag
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.asymmetric.ed25519 import (
    Ed25519PrivateKey,
    Ed25519PublicKey,
)
from cryptography.hazmat.primitives.asymmetric.x25519 import (
    X25519PrivateKey,
    X25519PublicKey,
)
from cryptography.hazmat.primitives.ciphers.aead import ChaCha20Poly1305
from cryptography.hazmat.primitives.kdf.hkdf import HKDF
from cryptography.hazmat.primitives.serialization import (
    Encoding,
    NoEncryption,
    PrivateFormat,
    PublicFormat,
)

from csam_guard import (
    ArmigerSafetyGate,
    CSAMGuard,
    CSAMViolationError,
    CSAMViolationType,
    ContentFilterResult,
    MLKEM768,
    SignedTombstoneList,
    TombstoneStore,
    WOTSPlusSHAKE256,
    blind_infohash,
    generate_ed25519_keypair,
    normalize_text,
    sanitize_title,
    sign_tombstone_list,
)

# Optional zstandard import with graceful zlib fallback
try:
    import zstandard as zstd  # type: ignore

    _HAS_ZSTD = True
except ImportError:
    _HAS_ZSTD = False


# ---------------------------------------------------------------------------
# Constants & Partition Taxonomies
# ---------------------------------------------------------------------------

CATEGORIES_MAINSTREAM: tuple[str, ...] = (
    "movie",
    "tv_show",
    "music",
    "game",
    "software",
    "anime",
    "book",
    "comic",
    "course",
)

CATEGORY_ADULT = "xxx"

ALL_CATEGORIES: tuple[str, ...] = CATEGORIES_MAINSTREAM + (CATEGORY_ADULT,)

LANGUAGES: tuple[str, ...] = (
    "en",
    "ja",
    "es",
    "fr",
    "de",
    "it",
    "pt",
    "ru",
    "zh",
    "ko",
    "hi",
    "ar",
    "multi",
    "other",
)

QUALITY_TIERS: tuple[str, ...] = (
    "uhd_4k",
    "hd_1080p",
    "hd_720p",
    "sd",
    "lossless_audio",
    "standard_audio",
    "any",
)

DEFAULT_EPOCH = "2026_q4"


class SyncMode(str, enum.Enum):
    """Operational mode for decentralized DHT sync."""

    OFF = "off"
    PULL_ONLY = "pull_only"
    SUBSCRIBE_AND_SEED = "subscribe_and_seed"
    PUBLISHER = "publisher"


# ---------------------------------------------------------------------------
# Data Models
# ---------------------------------------------------------------------------

@dataclasses.dataclass(frozen=True)
class TorrentRecord:
    """Canonical representation of an indexed torrent entry."""

    infohash: str  # 40-char hex
    title: str
    category: str
    language: str
    quality_tier: str
    epoch: str
    size_bytes: int
    file_count: int
    seeders: int
    leechers: int
    content_sha256: str
    file_list: list[str]
    safety_gate_passed: bool
    created_at: int

    def canonical_dict(self) -> dict[str, Any]:
        """Deterministic dictionary for serialization and hashing."""
        return {
            "category": self.category.strip().lower(),
            "content_sha256": self.content_sha256.strip().lower(),
            "created_at": int(self.created_at),
            "epoch": self.epoch.strip().lower(),
            "file_count": int(self.file_count),
            "file_list": sorted(self.file_list),
            "infohash": self.infohash.strip().lower(),
            "language": self.language.strip().lower(),
            "leechers": int(self.leechers),
            "quality_tier": self.quality_tier.strip().lower(),
            "safety_gate_passed": bool(self.safety_gate_passed),
            "seeders": int(self.seeders),
            "size_bytes": int(self.size_bytes),
            "title": self.title,
        }

    def canonical_bytes(self) -> bytes:
        return json.dumps(self.canonical_dict(), sort_keys=True, separators=(",", ":")).encode("utf-8")

    def leaf_hash(self) -> bytes:
        return hashlib.sha256(self.canonical_bytes()).digest()


@dataclasses.dataclass(frozen=True)
class QuorumAttestation:
    """Individual node attestation with Hybrid Ed25519 + WOTS+-SHAKE256 PQ signatures."""

    node_id: str  # 40-char hex DHT node ID
    asn: int  # Autonomous System Number (e.g. AS13335)
    ip_digest: str  # Blinded SHA256 of IP/subnet
    pubkey_hex: str  # Ed25519 public key (64 hex chars)
    signature_hex: str  # Ed25519 signature over manifest canonical hash
    timestamp: int
    pq_pubkey_hex: str = ""  # WOTS+-SHAKE256 PQ public key (64 hex chars)
    pq_signature_hex: str = ""  # WOTS+-SHAKE256 PQ signature (4352 hex chars)

    def canonical_dict(self) -> dict[str, Any]:
        return {
            "asn": int(self.asn),
            "ip_digest": self.ip_digest.lower(),
            "node_id": self.node_id.lower(),
            "pubkey_hex": self.pubkey_hex.lower(),
            "timestamp": int(self.timestamp),
        }

    def signing_payload(self, manifest_digest: bytes) -> bytes:
        meta = json.dumps(self.canonical_dict(), sort_keys=True, separators=(",", ":")).encode("utf-8")
        return hashlib.sha256(b"armiger-attestation-v2:" + manifest_digest + b":" + meta).digest()


@dataclasses.dataclass(frozen=True)
class DHTShardManifest:
    """BEP-44 / BEP-46 Hybrid Post-Quantum (Ed25519 + WOTS+-SHAKE256) signed manifest for an encrypted shard."""

    seq: int
    shard_id: str
    epoch: str
    record_count: int
    compressed_size_bytes: int
    uncompressed_size_bytes: int
    merkle_root: str  # 64-char hex
    ciphertext_sha256: str  # 64-char hex
    bep46_infohash: str  # 40-char hex
    timestamp: int
    root_pubkey_hex: str
    ed25519_signature_hex: str
    attestations: list[QuorumAttestation] = dataclasses.field(default_factory=list)
    root_pq_pubkey_hex: str = ""
    pq_signature_hex: str = ""

    def with_attestations(self, attestations: Sequence[QuorumAttestation]) -> DHTShardManifest:
        """Return a copy of the manifest with attached quorum attestations."""
        return dataclasses.replace(self, attestations=list(attestations))

    def canonical_dict_without_sig(self) -> dict[str, Any]:
        d: dict[str, Any] = {
            "bep46_infohash": self.bep46_infohash.lower(),
            "ciphertext_sha256": self.ciphertext_sha256.lower(),
            "compressed_size_bytes": int(self.compressed_size_bytes),
            "epoch": self.epoch.lower(),
            "merkle_root": self.merkle_root.lower(),
            "record_count": int(self.record_count),
            "root_pubkey_hex": self.root_pubkey_hex.lower(),
            "seq": int(self.seq),
            "shard_id": self.shard_id.lower(),
            "timestamp": int(self.timestamp),
            "uncompressed_size_bytes": int(self.uncompressed_size_bytes),
        }
        if self.root_pq_pubkey_hex:
            d["root_pq_pubkey_hex"] = self.root_pq_pubkey_hex.strip().lower()
        return d

    def canonical_payload(self) -> bytes:
        return json.dumps(self.canonical_dict_without_sig(), sort_keys=True, separators=(",", ":")).encode("utf-8")

    def canonical_digest(self) -> bytes:
        return hashlib.sha256(self.canonical_payload()).digest()

    def bep44_target(self, salt: str | None = None) -> str:
        """Compute BEP-44 DHT target lookup key: SHA1(pubkey || salt)."""
        pub_bytes = bytes.fromhex(self.root_pubkey_hex.strip().lower())
        salt_bytes = (salt.lower() if salt else f"armiger-shard:{self.shard_id.strip().lower()}").encode("utf-8")
        return hashlib.sha1(pub_bytes + salt_bytes).hexdigest()


# ---------------------------------------------------------------------------
# Merkle Tree Implementation
# ---------------------------------------------------------------------------

class MerkleTree:
    """Binary Merkle tree over canonical record leaf hashes with domain-separated odd-node padding."""

    ODD_NODE_SENTINEL: bytes = hashlib.sha256(b"armiger-merkle-odd-node-sentinel-v1").digest()

    def __init__(self, leaves: Sequence[bytes]) -> None:
        self.leaves = list(leaves)
        self.levels: list[list[bytes]] = []
        self._build()

    def _build(self) -> None:
        if not self.leaves:
            self.root = b"\x00" * 32
            self.levels = [[]]
            return

        current = self.leaves[:]
        self.levels.append(current)

        while len(current) > 1:
            next_level: list[bytes] = []
            for i in range(0, len(current), 2):
                left = current[i]
                right = current[i + 1] if i + 1 < len(current) else self.ODD_NODE_SENTINEL
                parent = hashlib.sha256(left + right).digest()
                next_level.append(parent)
            self.levels.append(next_level)
            current = next_level

        self.root = self.levels[-1][0]

    @property
    def root_hex(self) -> str:
        return self.root.hex()

    def get_proof(self, index: int) -> list[tuple[str, bool]]:
        """Return list of (sibling_hash_hex, is_right) proof pairs for leaf at index."""
        if index < 0 or index >= len(self.leaves):
            raise IndexError(f"Leaf index {index} out of range (total {len(self.leaves)})")

        proof: list[tuple[str, bool]] = []
        cur_idx = index
        for level in self.levels[:-1]:
            is_right = (cur_idx % 2 == 0)
            sib_idx = cur_idx + 1 if is_right else cur_idx - 1
            if sib_idx >= len(level):
                proof.append((self.ODD_NODE_SENTINEL.hex(), True))
            else:
                proof.append((level[sib_idx].hex(), is_right))
            cur_idx //= 2

        return proof

    @staticmethod
    def verify_proof(leaf_hash: bytes, proof: Sequence[tuple[str, bool]], root_hex: str) -> bool:
        """Verify an inclusion proof against expected Merkle root."""
        if not isinstance(leaf_hash, (bytes, bytearray)) or len(leaf_hash) != 32:
            return False
        try:
            expected_root = bytes.fromhex(str(root_hex).strip())
            if len(expected_root) != 32:
                return False
        except (ValueError, TypeError):
            return False

        current = bytes(leaf_hash)
        for item in proof:
            if not isinstance(item, (tuple, list)) or len(item) != 2:
                return False
            sib_hex, is_right = item
            try:
                sib = bytes.fromhex(str(sib_hex).strip())
            except (ValueError, TypeError):
                return False
            if len(sib) != 32:
                return False
            if is_right:
                current = hashlib.sha256(current + sib).digest()
            else:
                current = hashlib.sha256(sib + current).digest()
        return hmac.compare_digest(current, expected_root)


# ---------------------------------------------------------------------------
# X25519 & ChaCha20-Poly1305 AEAD Cryptographic Engine
# ---------------------------------------------------------------------------

class ShardCryptoEngine:
    """Post-Quantum Hybrid (X25519 + ML-KEM-768 + SHAKE-256 / HKDF-SHA3-256) & ChaCha20-Poly1305 AEAD Engine."""

    HYBRID_PRIV_LEN = 32 + MLKEM768.DK_BYTES  # 2,432 bytes
    HYBRID_PUB_LEN = 32 + MLKEM768.EK_BYTES   # 1,216 bytes
    HYBRID_ENC_LEN = 32 + MLKEM768.CT_BYTES   # 1,120 bytes

    @staticmethod
    def derive_shard_key(
        master_secret: bytes,
        shard_id: str,
        epoch: str,
        pq_shared_secret: bytes = b"",
    ) -> bytes:
        """Derive 256-bit ChaCha20-Poly1305 key via Dual-PRF (HKDF-SHA256 XOR SHAKE-256/SHA3-256)."""
        salt_bytes = shard_id.lower().encode("utf-8")
        info_bytes = f"armiger-shard-chacha20-poly1305-pq-v2:{epoch.lower()}".encode("utf-8")
        ikm = master_secret + pq_shared_secret
        k_hkdf = HKDF(
            algorithm=hashes.SHA256(),
            length=32,
            salt=salt_bytes,
            info=info_bytes,
        ).derive(ikm)
        k_shake = hashlib.shake_256(
            b"armiger-pq-shard-kdf-v2:" + salt_bytes + b":" + info_bytes + b":" + ikm
        ).digest(32)
        return bytes(a ^ b for a, b in zip(k_hkdf, k_shake))

    @classmethod
    def generate_pq_keypair(cls) -> tuple[bytes, bytes]:
        """Generate a Hybrid Post-Quantum (X25519 + ML-KEM-768) keypair: (priv_2432B, pub_1216B)."""
        priv = X25519PrivateKey.generate()
        pub = priv.public_key()
        x_priv = priv.private_bytes(Encoding.Raw, PrivateFormat.Raw, NoEncryption())
        x_pub = pub.public_bytes(Encoding.Raw, PublicFormat.Raw)
        pq_ek, pq_dk = MLKEM768.keygen()
        return x_priv + pq_dk, x_pub + pq_ek

    @classmethod
    def encapsulate_shard_key(
        cls,
        recipient_hybrid_pubkey: bytes,
        shard_id: str,
        epoch: str,
    ) -> tuple[bytes, bytes]:
        """Encapsulate a 256-bit shard key using Hybrid X25519 + ML-KEM-768, returning (kem_token_1120B, key_32B)."""
        if len(recipient_hybrid_pubkey) != cls.HYBRID_PUB_LEN:
            raise ValueError(f"Expected {cls.HYBRID_PUB_LEN}-byte hybrid public key")
        recip_x_pub = recipient_hybrid_pubkey[:32]
        recip_pq_ek = recipient_hybrid_pubkey[32:]
        ephem_priv = X25519PrivateKey.generate()
        ephem_x_pub = ephem_priv.public_key().public_bytes(Encoding.Raw, PublicFormat.Raw)
        ss_classical = ephem_priv.exchange(X25519PublicKey.from_public_bytes(recip_x_pub))
        pq_ct, ss_pq = MLKEM768.encaps(recip_pq_ek)
        kem_token = ephem_x_pub + pq_ct
        ctx = hashlib.sha3_256(kem_token).digest()
        key = cls.derive_shard_key(ss_classical, shard_id, epoch, pq_shared_secret=ss_pq + ctx)
        return kem_token, key

    @classmethod
    def decapsulate_shard_key(
        cls,
        recipient_hybrid_privkey: bytes,
        kem_token: bytes,
        shard_id: str,
        epoch: str,
    ) -> bytes:
        """Decapsulate a 1,120-byte X25519 + ML-KEM-768 token to recover the 256-bit shard key."""
        if len(recipient_hybrid_privkey) != cls.HYBRID_PRIV_LEN or len(kem_token) != cls.HYBRID_ENC_LEN:
            raise ValueError("Invalid hybrid private key or KEM encapsulation token length")
        x_priv = X25519PrivateKey.from_private_bytes(recipient_hybrid_privkey[:32])
        pq_dk = recipient_hybrid_privkey[32:]
        ephem_x_pub = kem_token[:32]
        pq_ct = kem_token[32:]
        ss_classical = x_priv.exchange(X25519PublicKey.from_public_bytes(ephem_x_pub))
        ss_pq = MLKEM768.decaps(pq_dk, pq_ct)
        ctx = hashlib.sha3_256(kem_token).digest()
        return cls.derive_shard_key(ss_classical, shard_id, epoch, pq_shared_secret=ss_pq + ctx)

    @staticmethod
    def construct_aad(shard_id: str, epoch: str, seq: int, merkle_root: str) -> bytes:
        """Deterministic Authenticated Associated Data (AAD) binding ciphertext to shard metadata."""
        return f"armiger-shard-aad-v1:{shard_id.lower()}:{epoch.lower()}:{seq}:{merkle_root.lower()}".encode("utf-8")

    @classmethod
    def encrypt_shard(
        cls,
        plaintext: bytes,
        key: bytes,
        shard_id: str,
        epoch: str,
        seq: int,
        merkle_root: str,
    ) -> bytes:
        """Encrypt shard payload with 256-bit ChaCha20-Poly1305 AEAD (128-bit Grover PQ security).

        Format: nonce (12 bytes) + ciphertext_with_poly1305_tag
        """
        if len(key) != 32:
            raise ValueError(f"ChaCha20 key must be 32 bytes, got {len(key)}")
        nonce = os.urandom(12)
        aad = cls.construct_aad(shard_id, epoch, seq, merkle_root)
        aead = ChaCha20Poly1305(key)
        ciphertext = aead.encrypt(nonce, plaintext, aad)
        return nonce + ciphertext

    @classmethod
    def decrypt_shard(
        cls,
        bundle: bytes,
        key: bytes,
        shard_id: str,
        epoch: str,
        seq: int,
        merkle_root: str,
    ) -> bytes:
        """Decrypt shard payload and verify Poly1305 tag against AAD.

        Raises InvalidTag if tampered or transplanted across different shard_id/epoch/seq.
        """
        if len(bundle) < 12 + 16:
            raise ValueError("Encrypted bundle is too short to contain nonce and tag")
        nonce = bundle[:12]
        ciphertext = bundle[12:]
        aad = cls.construct_aad(shard_id, epoch, seq, merkle_root)
        aead = ChaCha20Poly1305(key)
        return aead.decrypt(nonce, ciphertext, aad)

    @classmethod
    def encrypt_shard_pq(
        cls,
        plaintext: bytes,
        recipient_hybrid_pubkey: bytes,
        shard_id: str,
        epoch: str,
        seq: int,
        merkle_root: str,
    ) -> bytes:
        """Encrypt shard with ephemeral X25519 + ML-KEM-768 encapsulation + 256-bit ChaCha20-Poly1305 AEAD.

        Format: kem_token (1,120 bytes) + nonce (12 bytes) + ciphertext_with_poly1305_tag
        """
        kem_token, key = cls.encapsulate_shard_key(recipient_hybrid_pubkey, shard_id, epoch)
        nonce = os.urandom(12)
        aad = cls.construct_aad(shard_id, epoch, seq, merkle_root) + b":kem:" + hashlib.sha3_256(kem_token).digest()
        aead = ChaCha20Poly1305(key)
        ciphertext = aead.encrypt(nonce, plaintext, aad)
        return kem_token + nonce + ciphertext

    @classmethod
    def decrypt_shard_pq(
        cls,
        pq_bundle: bytes,
        recipient_hybrid_privkey: bytes,
        shard_id: str,
        epoch: str,
        seq: int,
        merkle_root: str,
    ) -> bytes:
        """Decrypt an X25519 + ML-KEM-768 + ChaCha20-Poly1305 shard bundle."""
        min_len = cls.HYBRID_ENC_LEN + 12 + 16
        if len(pq_bundle) < min_len:
            raise ValueError("PQ encrypted shard bundle too short")
        kem_token = pq_bundle[: cls.HYBRID_ENC_LEN]
        nonce = pq_bundle[cls.HYBRID_ENC_LEN : cls.HYBRID_ENC_LEN + 12]
        ciphertext = pq_bundle[cls.HYBRID_ENC_LEN + 12 :]
        key = cls.decapsulate_shard_key(recipient_hybrid_privkey, kem_token, shard_id, epoch)
        aad = cls.construct_aad(shard_id, epoch, seq, merkle_root) + b":kem:" + hashlib.sha3_256(kem_token).digest()
        aead = ChaCha20Poly1305(key)
        return aead.decrypt(nonce, ciphertext, aad)


# ---------------------------------------------------------------------------
# Compression Utilities (Zstd with Zlib fallback)
# ---------------------------------------------------------------------------

def compress_payload(data: bytes) -> tuple[bytes, str]:
    """Compress payload using Zstd (level 7) or fallback to Zlib (level 9)."""
    if _HAS_ZSTD:
        cctx = zstd.ZstdCompressor(level=7)
        return cctx.compress(data), "zstd"
    return zlib.compress(data, level=9), "zlib"


MAX_DECOMPRESSED_SHARD_BYTES = 4 * 1024 * 1024
MAX_DECOMPRESSION_RATIO = 50


def decompress_payload(
    compressed: bytes,
    codec: str = "zstd",
    max_output_bytes: int = MAX_DECOMPRESSED_SHARD_BYTES,
    max_ratio: int = MAX_DECOMPRESSION_RATIO,
) -> bytes:
    """Decompress payload using specified codec with strict zip-bomb size & ratio caps."""
    if not compressed:
        return b""
    hard_cap = min(max_output_bytes, max(65536, len(compressed) * max_ratio))

    def _zlib_bounded(data: bytes) -> bytes:
        dobj = zlib.decompressobj()
        out = dobj.decompress(data, hard_cap + 1)
        if dobj.unconsumed_tail or len(out) > hard_cap:
            return b""
        return out

    if codec == "zstd":
        if _HAS_ZSTD:
            try:
                dctx = zstd.ZstdDecompressor()
                out = dctx.decompress(compressed, max_output_size=hard_cap + 1)
                if len(out) > hard_cap:
                    return b""
                return out
            except Exception:
                return _zlib_bounded(compressed)
        return _zlib_bounded(compressed)
    elif codec == "zlib":
        return _zlib_bounded(compressed)
    raise ValueError(f"Unsupported compression codec: {codec}")


# ---------------------------------------------------------------------------
# Modular Sharded Subscription Config
# ---------------------------------------------------------------------------

@dataclasses.dataclass
class SubscriptionConfig:
    """User-selected subscription preferences for modular DHT sharding."""

    categories: set[str] = dataclasses.field(default_factory=lambda: {"movie", "tv_show"})
    languages: set[str] = dataclasses.field(default_factory=lambda: {"en", "ja"})
    quality_tiers: set[str] = dataclasses.field(default_factory=lambda: {"uhd_4k", "hd_1080p"})
    epochs: set[str] = dataclasses.field(default_factory=lambda: {DEFAULT_EPOCH})
    enable_xxx: bool = False  # Strictly isolated, default OFF
    min_quorum: int = 5
    min_distinct_asns: int = 5
    sync_mode: SyncMode = SyncMode.PULL_ONLY

    @classmethod
    def parse_shard_id(cls, shard_id: str) -> tuple[str, str, str, str]:
        """Deconstruct 'category/language/quality_tier/epoch' shard identifier."""
        parts = shard_id.strip().lower().split("/")
        if len(parts) != 4:
            raise ValueError(f"Malformed shard_id: '{shard_id}'. Expected 'category/lang/quality/epoch'")
        return parts[0], parts[1], parts[2], parts[3]

    @classmethod
    def format_shard_id(cls, category: str, language: str, quality_tier: str, epoch: str) -> str:
        return f"{category.lower()}/{language.lower()}/{quality_tier.lower()}/{epoch.lower()}"

    def is_subscribed(self, shard_id: str) -> bool:
        """Check whether a given shard matches user subscription filters.

        Invariant: If category == 'xxx' and enable_xxx is False, unconditionally REJECT.
        """
        try:
            cat, lang, quality, ep = self.parse_shard_id(shard_id)
        except ValueError:
            return False

        # Strict isolation gate for adult content
        if cat == CATEGORY_ADULT and not self.enable_xxx:
            return False

        return (
            cat in self.categories
            and lang in self.languages
            and quality in self.quality_tiers
            and ep in self.epochs
        )


# ---------------------------------------------------------------------------
# SQLite Columnar Shard Database Engine
# ---------------------------------------------------------------------------

class ShardDatabase:
    """Local SQLite database storing partitioned torrent metadata."""

    def __init__(self, db_path: str | pathlib.Path = ":memory:", enable_xxx: bool = False) -> None:
        self.db_path = str(db_path)
        self.enable_xxx = enable_xxx
        self.conn = sqlite3.connect(self.db_path)
        self.conn.row_factory = sqlite3.Row
        self._init_schema()

    def _init_schema(self) -> None:
        with self.conn:
            # Mainstream torrents table
            self.conn.execute("""
                CREATE TABLE IF NOT EXISTS mainstream_torrents (
                    infohash TEXT PRIMARY KEY,
                    title TEXT NOT NULL,
                    category TEXT NOT NULL,
                    language TEXT NOT NULL,
                    quality_tier TEXT NOT NULL,
                    epoch TEXT NOT NULL,
                    size_bytes INTEGER NOT NULL,
                    file_count INTEGER NOT NULL,
                    seeders INTEGER NOT NULL,
                    leechers INTEGER NOT NULL,
                    content_sha256 TEXT NOT NULL,
                    file_list_json TEXT NOT NULL,
                    created_at INTEGER NOT NULL
                )
            """)
            self.conn.execute("CREATE INDEX IF NOT EXISTS idx_main_cat_lang ON mainstream_torrents(category, language)")
            self.conn.execute("CREATE INDEX IF NOT EXISTS idx_main_epoch ON mainstream_torrents(epoch)")

            # Isolated adult table
            if self.enable_xxx:
                self.conn.execute("""
                    CREATE TABLE IF NOT EXISTS adult_torrents (
                        infohash TEXT PRIMARY KEY,
                        title TEXT NOT NULL,
                        category TEXT NOT NULL,
                        language TEXT NOT NULL,
                        quality_tier TEXT NOT NULL,
                        epoch TEXT NOT NULL,
                        size_bytes INTEGER NOT NULL,
                        file_count INTEGER NOT NULL,
                        seeders INTEGER NOT NULL,
                        leechers INTEGER NOT NULL,
                        content_sha256 TEXT NOT NULL,
                        file_list_json TEXT NOT NULL,
                        created_at INTEGER NOT NULL
                    )
                """)
                self.conn.execute("CREATE INDEX IF NOT EXISTS idx_adult_lang ON adult_torrents(language)")

    def insert_record(self, record: TorrentRecord) -> bool:
        """Insert a torrent record into the appropriate partitioned table."""
        cat_clean = record.category.strip().lower()
        if cat_clean not in ALL_CATEGORIES:
            return False
        is_adult = (cat_clean == CATEGORY_ADULT)
        if is_adult and not self.enable_xxx:
            return False  # Drop un-opted adult record

        table = "adult_torrents" if is_adult else "mainstream_torrents"
        with self.conn:
            self.conn.execute(
                f"""
                INSERT OR REPLACE INTO {table} (
                    infohash, title, category, language, quality_tier, epoch,
                    size_bytes, file_count, seeders, leechers, content_sha256,
                    file_list_json, created_at
                ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
                """,
                (
                    record.infohash.strip().lower(),
                    record.title,
                    cat_clean,
                    record.language.strip().lower(),
                    record.quality_tier.strip().lower(),
                    record.epoch.strip().lower(),
                    record.size_bytes,
                    record.file_count,
                    record.seeders,
                    record.leechers,
                    record.content_sha256.lower(),
                    json.dumps(record.file_list),
                    record.created_at,
                ),
            )
        return True

    def insert_records_batch(self, records: Sequence[TorrentRecord]) -> int:
        """Batch-insert validated torrent records within a single SQLite transaction."""
        if not records:
            return 0
        main_rows: list[tuple[Any, ...]] = []
        adult_rows: list[tuple[Any, ...]] = []
        for record in records:
            cat_clean = record.category.strip().lower()
            if cat_clean not in ALL_CATEGORIES:
                continue
            is_adult = (cat_clean == CATEGORY_ADULT)
            if is_adult and not self.enable_xxx:
                continue
            row = (
                record.infohash.strip().lower(),
                record.title,
                cat_clean,
                record.language.strip().lower(),
                record.quality_tier.strip().lower(),
                record.epoch.strip().lower(),
                record.size_bytes,
                record.file_count,
                record.seeders,
                record.leechers,
                record.content_sha256.lower(),
                json.dumps(record.file_list),
                record.created_at,
            )
            if is_adult:
                adult_rows.append(row)
            else:
                main_rows.append(row)

        with self.conn:
            if main_rows:
                self.conn.executemany(
                    """
                    INSERT OR REPLACE INTO mainstream_torrents (
                        infohash, title, category, language, quality_tier, epoch,
                        size_bytes, file_count, seeders, leechers, content_sha256,
                        file_list_json, created_at
                    ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
                    """,
                    main_rows,
                )
            if adult_rows and self.enable_xxx:
                self.conn.executemany(
                    """
                    INSERT OR REPLACE INTO adult_torrents (
                        infohash, title, category, language, quality_tier, epoch,
                        size_bytes, file_count, seeders, leechers, content_sha256,
                        file_list_json, created_at
                    ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
                    """,
                    adult_rows,
                )
        return len(main_rows) + len(adult_rows)

    def get_record(self, infohash: str) -> TorrentRecord | None:
        """Retrieve record by infohash across accessible tables."""
        clean = infohash.lower().strip()
        cur = self.conn.cursor()
        cur.execute("SELECT * FROM mainstream_torrents WHERE infohash = ?", (clean,))
        row = cur.fetchone()
        if row is None and self.enable_xxx:
            cur.execute("SELECT * FROM adult_torrents WHERE infohash = ?", (clean,))
            row = cur.fetchone()

        if row is None:
            return None

        return TorrentRecord(
            infohash=row["infohash"],
            title=row["title"],
            category=row["category"],
            language=row["language"],
            quality_tier=row["quality_tier"],
            epoch=row["epoch"],
            size_bytes=row["size_bytes"],
            file_count=row["file_count"],
            seeders=row["seeders"],
            leechers=row["leechers"],
            content_sha256=row["content_sha256"],
            file_list=json.loads(row["file_list_json"]),
            safety_gate_passed=True,
            created_at=row["created_at"],
        )

    def purge_blinded_hashes(self, tombstone_store: TombstoneStore) -> int:
        """Purge any records whose blinded infohashes are registered on tombstone list."""
        purged = 0
        cur = self.conn.cursor()

        # Check mainstream
        cur.execute("SELECT infohash FROM mainstream_torrents")
        main_hashes = [r[0] for r in cur.fetchall()]
        to_delete_main = [h for h in main_hashes if tombstone_store.is_tombstoned(h)]
        if to_delete_main:
            with self.conn:
                self.conn.executemany("DELETE FROM mainstream_torrents WHERE infohash = ?", [(h,) for h in to_delete_main])
            purged += len(to_delete_main)

        # Check adult if enabled
        if self.enable_xxx:
            cur.execute("SELECT infohash FROM adult_torrents")
            adult_hashes = [r[0] for r in cur.fetchall()]
            to_delete_adult = [h for h in adult_hashes if tombstone_store.is_tombstoned(h)]
            if to_delete_adult:
                with self.conn:
                    self.conn.executemany("DELETE FROM adult_torrents WHERE infohash = ?", [(h,) for h in to_delete_adult])
                purged += len(to_delete_adult)

        return purged

    def count_records(self, category: str | None = None) -> int:
        cur = self.conn.cursor()
        if category == CATEGORY_ADULT:
            if not self.enable_xxx:
                return 0
            cur.execute("SELECT COUNT(*) FROM adult_torrents")
            return int(cur.fetchone()[0])
        elif category is not None:
            cur.execute("SELECT COUNT(*) FROM mainstream_torrents WHERE category = ?", (category.lower(),))
            return int(cur.fetchone()[0])
        else:
            cur.execute("SELECT COUNT(*) FROM mainstream_torrents")
            tot = int(cur.fetchone()[0])
            if self.enable_xxx:
                cur.execute("SELECT COUNT(*) FROM adult_torrents")
                tot += int(cur.fetchone()[0])
            return tot

    def search(self, query: str = "", category: str | None = None, language: str | None = None) -> list[TorrentRecord]:
        """Search records by title query, category, and language."""
        cur = self.conn.cursor()
        is_adult_query = (category == CATEGORY_ADULT)
        if is_adult_query and not self.enable_xxx:
            return []

        table = "adult_torrents" if is_adult_query else "mainstream_torrents"
        sql = f"SELECT * FROM {table} WHERE 1=1"
        params: list[Any] = []

        if query:
            sql += " AND title LIKE ?"
            params.append(f"%{query}%")
        if category and not is_adult_query:
            sql += " AND category = ?"
            params.append(category.lower())
        if language:
            sql += " AND language = ?"
            params.append(language.lower())

        cur.execute(sql, params)
        results: list[TorrentRecord] = []
        for row in cur.fetchall():
            results.append(TorrentRecord(
                infohash=row["infohash"],
                title=row["title"],
                category=row["category"],
                language=row["language"],
                quality_tier=row["quality_tier"],
                epoch=row["epoch"],
                size_bytes=row["size_bytes"],
                file_count=row["file_count"],
                seeders=row["seeders"],
                leechers=row["leechers"],
                content_sha256=row["content_sha256"],
                file_list=json.loads(row["file_list_json"]),
                safety_gate_passed=True,
                created_at=row["created_at"],
            ))
        return results

    def close(self) -> None:
        self.conn.close()

    def __del__(self) -> None:
        try:
            self.close()
        except Exception:
            pass


# ---------------------------------------------------------------------------
# Shard Serialization & Deserialization
# ---------------------------------------------------------------------------

def serialize_shard_records(records: Sequence[TorrentRecord]) -> tuple[bytes, MerkleTree]:
    """Serialize a list of TorrentRecords into a deterministic compressed micro-bundle."""
    # Deterministic sorting by infohash
    sorted_records = sorted(records, key=lambda r: r.infohash.lower())
    leaves = [r.leaf_hash() for r in sorted_records]
    tree = MerkleTree(leaves)

    payload_dict = {
        "format": "armiger-shard-v1",
        "record_count": len(sorted_records),
        "records": [r.canonical_dict() for r in sorted_records],
    }
    raw_json = json.dumps(payload_dict, sort_keys=True, separators=(",", ":")).encode("utf-8")
    compressed, _codec = compress_payload(raw_json)
    return compressed, tree


def deserialize_shard_records(compressed: bytes) -> list[TorrentRecord]:
    """Decompress and parse shard records."""
    raw_json = decompress_payload(compressed)
    payload_dict = json.loads(raw_json.decode("utf-8"))
    records: list[TorrentRecord] = []
    for d in payload_dict.get("records", []):
        records.append(TorrentRecord(
            infohash=d["infohash"],
            title=d["title"],
            category=d["category"],
            language=d["language"],
            quality_tier=d["quality_tier"],
            epoch=d["epoch"],
            size_bytes=d["size_bytes"],
            file_count=d["file_count"],
            seeders=d["seeders"],
            leechers=d["leechers"],
            content_sha256=d["content_sha256"],
            file_list=d.get("file_list", []),
            safety_gate_passed=bool(d.get("safety_gate_passed", True)),
            created_at=d.get("created_at", int(time.time())),
        ))
    return records


# ---------------------------------------------------------------------------
# Quorum & Attestation Verification Engine
# ---------------------------------------------------------------------------

class QuorumVerifier:
    """Validates K-of-N distinct-ASN quorum attestations and Hybrid Ed25519 + WOTS+-SHAKE256 root signatures."""

    def __init__(
        self,
        trusted_root_pubkeys_hex: Sequence[str],
        min_quorum: int = 5,
        min_distinct_asns: int = 5,
        trusted_root_pq_pubkeys_hex: Sequence[str] | None = None,
    ) -> None:
        self.trusted_roots = [
            bytes.fromhex(k.strip().lower())
            for k in trusted_root_pubkeys_hex
            if len(k.strip()) == 64
        ]
        self.trusted_pq_roots = [
            bytes.fromhex(k.strip().lower())
            for k in (trusted_root_pq_pubkeys_hex or [])
            if len(k.strip()) == 64
        ]
        self.min_quorum = min_quorum
        self.min_distinct_asns = min_distinct_asns

    def verify_manifest(
        self,
        manifest: DHTShardManifest,
        current_seq: int = -1,
    ) -> tuple[bool, str]:
        """Verify:
          1. Monotonic sequence: seq > current_seq
          2. Root Hybrid Ed25519 + WOTS+-SHAKE256 Post-Quantum signature
          3. K-of-N attestation count >= min_quorum
          4. Distinct ASNs >= min_distinct_asns
          5. Each attestation Hybrid Ed25519 + WOTS+-SHAKE256 signature
        """
        # 0. Policy & type sanity checks
        if isinstance(manifest.seq, bool) or not isinstance(manifest.seq, int) or manifest.seq < 0:
            return False, "invalid_manifest_seq_type"
        if self.min_quorum <= 0 or self.min_distinct_asns <= 0:
            return False, "invalid_quorum_policy_threshold"
        if not self.trusted_roots:
            return False, "no_trusted_root_pubkeys_configured"

        # 1. Monotonic sequence check
        if manifest.seq <= current_seq:
            return False, f"stale_or_downgrade_sequence:{manifest.seq}<={current_seq}"

        # 2. Root Ed25519 signature check
        try:
            root_pub_bytes = bytes.fromhex(manifest.root_pubkey_hex.strip().lower())
        except ValueError:
            return False, "invalid_root_pubkey_hex"
        if root_pub_bytes not in self.trusted_roots:
            return False, f"untrusted_root_pubkey:{manifest.root_pubkey_hex}"

        payload = manifest.canonical_payload()
        try:
            vk = Ed25519PublicKey.from_public_bytes(root_pub_bytes)
            vk.verify(bytes.fromhex(manifest.ed25519_signature_hex.strip()), payload)
        except (InvalidSignature, ValueError) as exc:
            return False, f"invalid_root_ed25519_signature:{exc}"

        # 2b. Root WOTS+-SHAKE256 Post-Quantum signature check
        if manifest.pq_signature_hex or manifest.root_pq_pubkey_hex or self.trusted_pq_roots:
            try:
                pq_root_bytes = bytes.fromhex(manifest.root_pq_pubkey_hex.strip().lower())
                pq_sig_bytes = bytes.fromhex(manifest.pq_signature_hex.strip())
            except ValueError as exc:
                return False, f"invalid_root_pq_wots_signature:{exc}"
            if self.trusted_pq_roots and pq_root_bytes not in self.trusted_pq_roots:
                return False, f"untrusted_root_pq_pubkey:{manifest.root_pq_pubkey_hex}"
            if not WOTSPlusSHAKE256.verify(pq_root_bytes, payload, pq_sig_bytes):
                return False, "invalid_root_pq_wots_signature:verification_failed"

        # 3. Quorum threshold check
        if len(manifest.attestations) < self.min_quorum:
            return False, f"insufficient_attestations:{len(manifest.attestations)}<{self.min_quorum}"

        # 4. Validate individual attestations & count distinct ASNs
        manifest_digest = manifest.canonical_digest()
        verified_asns: set[int] = set()
        seen_node_ids: set[str] = set()
        seen_pubkeys: set[str] = set()
        seen_pq_pubkeys: set[str] = set()
        seen_sigs: set[str] = set()
        seen_ip_digests: set[str] = set()
        now = int(time.time())

        for att in manifest.attestations:
            norm_node = str(att.node_id).strip().lower()
            norm_pk = str(att.pubkey_hex).strip().lower()
            norm_sig = str(att.signature_hex).strip().lower()
            norm_ip = str(att.ip_digest).strip().lower()

            # Reject duplicate node_id, pubkey, signature, or IP digest Sybil replay
            if norm_pk in seen_pubkeys or norm_sig in seen_sigs:
                return False, f"sybil_duplicate_attestor_pubkey_or_signature:{norm_pk[:16]}"
            if norm_node in seen_node_ids or (norm_ip and norm_ip in seen_ip_digests):
                continue

            seen_node_ids.add(norm_node)
            seen_pubkeys.add(norm_pk)
            seen_sigs.add(norm_sig)
            if norm_ip:
                seen_ip_digests.add(norm_ip)

            if att.pq_pubkey_hex:
                norm_pq_pk = str(att.pq_pubkey_hex).strip().lower()
                if norm_pq_pk in seen_pq_pubkeys:
                    return False, f"sybil_duplicate_attestor_pq_pubkey:{norm_pq_pk[:16]}"
                seen_pq_pubkeys.add(norm_pq_pk)

            # Enforce attestation timestamp freshness (reject ts<=0, >24h stale, or >5m in future)
            if isinstance(att.timestamp, bool) or not isinstance(att.timestamp, int):
                continue
            if att.timestamp <= 0 or (now - att.timestamp) > 86400 or (att.timestamp - now) > 300:
                continue

            if isinstance(att.asn, bool) or not isinstance(att.asn, int) or att.asn <= 0:
                continue

            bound_digest = att.signing_payload(manifest_digest)
            sig_msg_used = bound_digest
            try:
                node_vk = Ed25519PublicKey.from_public_bytes(bytes.fromhex(norm_pk))
                try:
                    node_vk.verify(bytes.fromhex(norm_sig), bound_digest)
                except InvalidSignature:
                    node_vk.verify(bytes.fromhex(norm_sig), manifest_digest)
                    sig_msg_used = manifest_digest
            except (InvalidSignature, ValueError):
                continue

            if att.pq_signature_hex or att.pq_pubkey_hex:
                try:
                    node_pq_pk = bytes.fromhex(att.pq_pubkey_hex.strip().lower())
                    node_pq_sig = bytes.fromhex(att.pq_signature_hex.strip())
                    if not (
                        WOTSPlusSHAKE256.verify(node_pq_pk, sig_msg_used, node_pq_sig)
                        or WOTSPlusSHAKE256.verify(node_pq_pk, manifest_digest, node_pq_sig)
                    ):
                        continue
                except ValueError:
                    continue

            verified_asns.add(int(att.asn))

        if len(verified_asns) < self.min_distinct_asns:
            return False, f"insufficient_distinct_asns:{len(verified_asns)}<{self.min_distinct_asns}"

        return True, "quorum_verified"


# ---------------------------------------------------------------------------
# Main Plugin Interface: ArmigerDHTDatabaseSync
# ---------------------------------------------------------------------------

class ArmigerDHTDatabaseSync:
    """End-to-end decentralized, CSAM-safe, encrypted, modular DHT database sync plugin."""

    CATEGORY_TO_CODE: dict[str, int] = {
        cat: idx for idx, cat in enumerate(ALL_CATEGORIES)
    }

    def __init__(
        self,
        subscription: SubscriptionConfig | None = None,
        master_secret: bytes | None = None,
        trusted_root_pubkeys_hex: Sequence[str] | None = None,
        db_path: str | pathlib.Path = ":memory:",
        tombstone_store: TombstoneStore | None = None,
        *,
        opt_in_sharing: bool | None = None,
        ephemeral_identity: bool = True,
        min_k_samples: int = 25,
    ) -> None:
        self.opt_in_sharing: bool = resolve_opt_in_sharing(opt_in_sharing)
        self.subscription = subscription or SubscriptionConfig(
            sync_mode=SyncMode.SUBSCRIBE_AND_SEED if self.opt_in_sharing else SyncMode.PULL_ONLY
        )
        self.master_secret = master_secret or os.urandom(32)
        self.tombstone_store = tombstone_store or TombstoneStore(trusted_root_pubkeys_hex)
        self.csam_guard = CSAMGuard(self.tombstone_store)
        self.safety_gate = ArmigerSafetyGate()
        self.quorum_verifier = QuorumVerifier(
            trusted_root_pubkeys_hex=trusted_root_pubkeys_hex or [],
            min_quorum=self.subscription.min_quorum,
            min_distinct_asns=self.subscription.min_distinct_asns,
        )
        self.db = ShardDatabase(db_path=db_path, enable_xxx=self.subscription.enable_xxx)
        self.shard_sequences: dict[str, int] = {}
        self.anonymity_guard = AnonymityGuard(
            ephemeral_identity=ephemeral_identity,
            min_k_samples=min_k_samples,
        )
        self._outbound_bundles_shared: int = 0
        self._outbound_packets_blocked_opt_out: int = 0
        self._outbound_bytes_shared: int = 0
        self._outbound_seq: int = 0

    def set_sync_mode(self, mode: SyncMode) -> None:
        self.subscription.sync_mode = mode

    def ingest_candidate_record(
        self,
        infohash: str,
        title: str,
        category: str,
        language: str,
        quality_tier: str,
        epoch: str = DEFAULT_EPOCH,
        size_bytes: int = 0,
        file_count: int = 1,
        seeders: int = 0,
        leechers: int = 0,
        content_sha256: str = "",
        file_list: list[str] | None = None,
        p_adult: float = 0.0,
        p_junk: float = 0.0,
        p_excluded_scope: float = 0.0,
    ) -> tuple[bool, str, TorrentRecord | None]:
        """Validate candidate record through CSAM guard and safety gates, then insert locally."""
        if self.subscription.sync_mode == SyncMode.OFF:
            return False, "sync_mode_off", None

        # 1. CSAM & structural harmful content blocker
        files = file_list or [title]
        res = self.csam_guard.inspect_torrent(
            infohash=infohash,
            title=title,
            file_list=files,
            category=category,
        )
        if not res.is_safe:
            return False, f"csam_guard_rejection:{res.violation_type.value}:{res.reason}", None

        # 2. Armiger Conformal Safety Gate
        gate_ok, gate_reason, gate_violation = self.safety_gate.evaluate(
            p_adult=p_adult,
            p_junk=p_junk,
            p_excluded_scope=p_excluded_scope,
            target_category=category,
        )
        if not gate_ok:
            return False, f"safety_gate_rejection:{gate_reason}", None

        # 3. Construct canonical TorrentRecord
        clean_sha = content_sha256 or hashlib.sha256(title.encode()).hexdigest()
        record = TorrentRecord(
            infohash=infohash.lower().strip(),
            title=res.sanitized_title,
            category=category.lower().strip(),
            language=language.lower().strip(),
            quality_tier=quality_tier.lower().strip(),
            epoch=epoch.lower().strip(),
            size_bytes=size_bytes,
            file_count=file_count,
            seeders=seeders,
            leechers=leechers,
            content_sha256=clean_sha.lower(),
            file_list=files,
            safety_gate_passed=True,
            created_at=int(time.time()),
        )

        # 4. Insert into local partitioned database
        inserted = self.db.insert_record(record)
        if not inserted:
            return False, "record_dropped_by_subscription_policy", None

        return True, "ingested", record

    def build_and_sign_shard(
        self,
        shard_id: str,
        records: Sequence[TorrentRecord],
        root_privkey: bytes,
        seq: int,
        attestations: Sequence[QuorumAttestation] | None = None,
    ) -> tuple[DHTShardManifest, bytes]:
        """Package a list of records into an encrypted micro-shard and generate signed manifest."""
        cat, lang, qual, epoch = SubscriptionConfig.parse_shard_id(shard_id)

        # 1. Serialize records and compute Merkle tree
        compressed_records, merkle_tree = serialize_shard_records(records)
        uncompressed_size = sum(len(r.canonical_bytes()) for r in records)

        # 2. Derive ChaCha20-Poly1305 key & encrypt
        key = ShardCryptoEngine.derive_shard_key(self.master_secret, shard_id, epoch)
        ciphertext_bundle = ShardCryptoEngine.encrypt_shard(
            plaintext=compressed_records,
            key=key,
            shard_id=shard_id,
            epoch=epoch,
            seq=seq,
            merkle_root=merkle_tree.root_hex,
        )

        ciphertext_sha = hashlib.sha256(ciphertext_bundle).hexdigest()
        bep46_infohash = hashlib.sha1(ciphertext_bundle).hexdigest()

        # 3. Create manifest and sign with Ed25519 + WOTS+-SHAKE256 root private key
        sk = Ed25519PrivateKey.from_private_bytes(root_privkey)
        pk = sk.public_key()
        root_pub_bytes = pk.public_bytes(Encoding.Raw, PublicFormat.Raw)
        pq_priv, pq_pub = WOTSPlusSHAKE256.derive_keypair_from_ed25519_priv(root_privkey)

        manifest_dummy = DHTShardManifest(
            seq=seq,
            shard_id=shard_id,
            epoch=epoch,
            record_count=len(records),
            compressed_size_bytes=len(ciphertext_bundle),
            uncompressed_size_bytes=uncompressed_size,
            merkle_root=merkle_tree.root_hex,
            ciphertext_sha256=ciphertext_sha,
            bep46_infohash=bep46_infohash,
            timestamp=int(time.time()),
            root_pubkey_hex=root_pub_bytes.hex(),
            ed25519_signature_hex="",
            attestations=list(attestations or []),
            root_pq_pubkey_hex=pq_pub.hex(),
        )

        payload_bytes = manifest_dummy.canonical_payload()
        sig = sk.sign(payload_bytes)
        pq_sig = WOTSPlusSHAKE256.sign(pq_priv, payload_bytes)

        manifest = DHTShardManifest(
            seq=seq,
            shard_id=shard_id,
            epoch=epoch,
            record_count=len(records),
            compressed_size_bytes=len(ciphertext_bundle),
            uncompressed_size_bytes=uncompressed_size,
            merkle_root=merkle_tree.root_hex,
            ciphertext_sha256=ciphertext_sha,
            bep46_infohash=bep46_infohash,
            timestamp=manifest_dummy.timestamp,
            root_pubkey_hex=root_pub_bytes.hex(),
            ed25519_signature_hex=sig.hex(),
            attestations=list(attestations or []),
            root_pq_pubkey_hex=pq_pub.hex(),
            pq_signature_hex=pq_sig.hex(),
        )

        return manifest, ciphertext_bundle

    def receive_and_apply_shard(
        self,
        manifest: DHTShardManifest,
        ciphertext_bundle: bytes,
    ) -> tuple[bool, str, int]:
        """Validate, decrypt, Merkle-verify, and ingest an incoming decentralized shard."""
        if self.subscription.sync_mode == SyncMode.OFF:
            return False, "sync_mode_off", 0

        # 1. Check user subscription filter (zero bandwidth / zero storage for unselected shards)
        if not self.subscription.is_subscribed(manifest.shard_id):
            return False, f"unsubscribed_shard_filtered:{manifest.shard_id}", 0

        # 2. Verify monotonic sequence & quorum
        cur_seq = self.shard_sequences.get(manifest.shard_id, -1)
        q_ok, q_msg = self.quorum_verifier.verify_manifest(manifest, current_seq=cur_seq)
        if not q_ok:
            return False, f"quorum_verification_failed:{q_msg}", 0

        # 3. Content-addressed & size verification of ciphertext
        actual_sha = hashlib.sha256(ciphertext_bundle).hexdigest()
        if actual_sha.lower() != manifest.ciphertext_sha256.lower():
            return False, "ciphertext_sha256_mismatch", 0
        if len(ciphertext_bundle) != manifest.compressed_size_bytes:
            return False, f"compressed_size_mismatch:{len(ciphertext_bundle)}!={manifest.compressed_size_bytes}", 0

        # 4. Decrypt shard with AEAD
        key = ShardCryptoEngine.derive_shard_key(self.master_secret, manifest.shard_id, manifest.epoch)
        try:
            plaintext = ShardCryptoEngine.decrypt_shard(
                bundle=ciphertext_bundle,
                key=key,
                shard_id=manifest.shard_id,
                epoch=manifest.epoch,
                seq=manifest.seq,
                merkle_root=manifest.merkle_root,
            )
        except (InvalidTag, ValueError) as exc:
            return False, f"aead_decryption_failed:{exc}", 0

        # 5. Decompress records & verify Merkle root and declared counts/sizes
        try:
            records = deserialize_shard_records(plaintext)
        except Exception as exc:
            return False, f"shard_deserialization_failed:{exc}", 0

        if len(records) != manifest.record_count:
            return False, f"record_count_mismatch:{len(records)}!={manifest.record_count}", 0
        actual_uncomp = sum(len(r.canonical_bytes()) for r in records)
        if actual_uncomp != manifest.uncompressed_size_bytes:
            return False, f"uncompressed_size_mismatch:{actual_uncomp}!={manifest.uncompressed_size_bytes}", 0
        reconstructed_leaves = [r.leaf_hash() for r in sorted(records, key=lambda r: r.infohash.lower())]
        reconstructed_tree = MerkleTree(reconstructed_leaves)

        if reconstructed_tree.root_hex.lower() != manifest.merkle_root.lower():
            return False, "merkle_root_mismatch", 0

        # 6. Pre-ingest safety re-inspection of decrypted records
        accepted_records: list[TorrentRecord] = []
        for rec in records:
            # Check against tombstone store
            if self.tombstone_store.is_tombstoned(rec.infohash):
                continue

            # CSAM inspection
            c_res = self.csam_guard.inspect_torrent(
                infohash=rec.infohash,
                title=rec.title,
                file_list=rec.file_list,
                category=rec.category,
            )
            if not c_res.is_safe:
                continue

            accepted_records.append(rec)

        ingested_count = self.db.insert_records_batch(accepted_records)

        # 7. Update sequence number
        self.shard_sequences[manifest.shard_id] = manifest.seq
        return True, "shard_applied_successfully", ingested_count

    def apply_signed_tombstones(self, manifest: SignedTombstoneList) -> tuple[bool, str, int, int]:
        """Apply a signed tombstone list and purge local records.

        Returns (ok, msg, new_tombstones_count, purged_records_count).
        """
        ok, msg, count = self.tombstone_store.apply_signed_tombstones(manifest)
        if not ok:
            return False, msg, 0, 0

        purged = self.db.purge_blinded_hashes(self.tombstone_store)
        return True, msg, count, purged

    def sharing_status(self) -> dict[str, Any]:
        """Return runtime audit telemetry confirming opt-in status and outbound packet/bundle counters."""
        return {
            "opt_in_sharing_enabled": bool(self.opt_in_sharing),
            "sync_mode": self.subscription.sync_mode.value,
            "outbound_bundles_shared": int(self._outbound_bundles_shared),
            "outbound_packets_blocked_opt_out": int(self._outbound_packets_blocked_opt_out),
            "outbound_bytes_shared": int(self._outbound_bytes_shared),
            "ephemeral_identity_enabled": bool(self.anonymity_guard.ephemeral_identity),
            "ephemeral_pubkey_hex": self.anonymity_guard.current_pubkey_hex() if self.opt_in_sharing else None,
            "ephemeral_rotations": int(self.anonymity_guard.rotation_count),
            "blocked_anonymity_violations": int(self.anonymity_guard.blocked_violations_count),
        }

    def hash_catalog_key(self, normalized_key: str, *, reject_raw_pii: bool = True) -> str:
        """Compute strictly 16-hex-char truncated BLAKE2b-64(domain='armiger-catalog-v2' || normalized_key)."""
        return self.anonymity_guard.hash_catalog_key(normalized_key, reject_raw_pii=reject_raw_pii)

    def export_signed_bundle(
        self,
        records: Sequence[Mapping[str, Any] | Sequence[Any]] | None = None,
        *,
        epoch: int | float | None = None,
        seq: int | None = None,
        n_samples: int = 25,
        salt: bytes = b"armiger-catalog-v2",
        raw_wire_override: bytes | str | Mapping[str, Any] | None = None,
        rotate_key: bool | None = None,
        as_dict_records: bool = True,
    ) -> dict[str, Any] | None:
        """Export an anonymous BEP-44 catalog record bundle with 16-hex BLAKE2b keys `[h, c, r, s, a, v, n, t]`.

        Strictly returns None (0 bytes shared) when `opt_in_sharing=False`.
        Raises `AnonymityViolationError` if any PII/canary/schema/k-anonymity invariant is violated.
        """
        if not self.opt_in_sharing:
            self._outbound_packets_blocked_opt_out += 1
            return None

        self._outbound_seq += 1
        next_seq = int(seq) if seq is not None else self._outbound_seq

        if raw_wire_override is not None:
            if isinstance(raw_wire_override, (bytes, bytearray)):
                v_bytes = bytes(raw_wire_override)
            elif isinstance(raw_wire_override, str):
                v_bytes = raw_wire_override.encode("utf-8")
            elif isinstance(raw_wire_override, Mapping):
                v_bytes = json.dumps(dict(raw_wire_override), sort_keys=True, separators=(",", ":")).encode("utf-8")
            else:
                raise AnonymityViolationError("invalid_raw_wire_override")
            signed = self.anonymity_guard.sign_wire_payload(
                v_bytes, seq=next_seq, salt=salt, rotate_key=rotate_key
            )
            self._outbound_bundles_shared += 1
            self._outbound_bytes_shared += len(v_bytes)
            return signed

        if isinstance(n_samples, bool) or not isinstance(n_samples, int) or n_samples < self.anonymity_guard.min_k_samples:
            self.anonymity_guard.blocked_violations_count += 1
            raise AnonymityViolationError(f"k_anonymity_violation:n_samples={n_samples}<{self.anonymity_guard.min_k_samples}")

        epoch_bucket = self.anonymity_guard.bucket_timestamp(epoch)
        sanitized_records: list[Any] = []
        for rec in (records or []):
            if isinstance(rec, Mapping):
                extra = set(rec.keys()) - {
                    "h", "key", "normalized_key", "c", "category", "r", "s", "a", "v", "n", "t"
                }
                if extra:
                    self.anonymity_guard.blocked_violations_count += 1
                    raise AnonymityViolationError(f"unexpected_catalog_record_keys:{sorted(extra)}")
                raw_h = rec.get("h")
                raw_key = rec.get("key") or rec.get("normalized_key")
                if raw_h is not None:
                    h_str = str(raw_h)
                    is_pre = True
                elif raw_key is not None:
                    h_str = str(raw_key)
                    is_pre = False
                else:
                    self.anonymity_guard.blocked_violations_count += 1
                    raise AnonymityViolationError("missing_catalog_key_or_hash")
                cat_in = rec.get("c", rec.get("category", 0))
                c_code = self.CATEGORY_TO_CODE.get(str(cat_in).lower(), 0) if isinstance(cat_in, str) else int(cat_in)
                r_code = int(rec.get("r", 0))
                s_code = int(rec.get("s", 0))
                a_code = 1 if rec.get("a", 0) else 0
                v_code = rec.get("v", 1)
                n_cnt = int(rec.get("n", n_samples))
                t_ts = rec.get("t", epoch_bucket)
                if isinstance(t_ts, int) and t_ts % self.anonymity_guard.bucket_seconds != 0:
                    self.anonymity_guard.blocked_violations_count += 1
                    raise AnonymityViolationError(f"record_unbucketed_timestamp:t={t_ts}")
                built = self.anonymity_guard.build_anonymous_catalog_record(
                    key_or_hash=h_str,
                    category_code=c_code,
                    res_bucket=r_code,
                    size_bucket=s_code,
                    adult_flag=a_code,
                    viability_bucket=v_code,
                    sample_count=n_cnt,
                    timestamp=t_ts,
                    is_prehashed=is_pre,
                    as_dict=as_dict_records,
                )
                sanitized_records.append(built)
            elif isinstance(rec, (list, tuple)) and len(rec) == 8:
                h_val, c_val, r_val, s_val, a_val, v_val, n_val, t_val = rec
                if isinstance(t_val, int) and t_val % self.anonymity_guard.bucket_seconds != 0:
                    self.anonymity_guard.blocked_violations_count += 1
                    raise AnonymityViolationError(f"record_unbucketed_timestamp:t={t_val}")
                built = self.anonymity_guard.build_anonymous_catalog_record(
                    key_or_hash=str(h_val),
                    category_code=int(c_val),
                    res_bucket=int(r_val),
                    size_bucket=int(s_val),
                    adult_flag=int(a_val),
                    viability_bucket=v_val,
                    sample_count=int(n_val),
                    timestamp=int(t_val),
                    is_prehashed=True,
                    as_dict=as_dict_records,
                )
                sanitized_records.append(built)
            else:
                self.anonymity_guard.blocked_violations_count += 1
                raise AnonymityViolationError(f"malformed_catalog_record:{rec!r}")

        if not sanitized_records:
            default_h = self.hash_catalog_key("default|2026")
            sanitized_records.append(
                self.anonymity_guard.build_anonymous_catalog_record(
                    key_or_hash=default_h,
                    category_code=0,
                    sample_count=n_samples,
                    timestamp=epoch_bucket,
                    is_prehashed=True,
                    as_dict=as_dict_records,
                )
            )

        wire_obj = {
            "v": 2,
            "epoch": epoch_bucket,
            "seq": next_seq,
            "n_bucket": (int(n_samples) // 25) * 25,
            "records": sanitized_records,
        }
        v_bytes = json.dumps(wire_obj, sort_keys=True, separators=(",", ":")).encode("utf-8")
        signed = self.anonymity_guard.sign_wire_payload(
            v_bytes, seq=next_seq, salt=salt, rotate_key=rotate_key
        )
        self._outbound_bundles_shared += 1
        self._outbound_bytes_shared += len(v_bytes)
        return signed

    def export_signed_delta_bundle(self, *args: Any, **kwargs: Any) -> dict[str, Any] | None:
        """Alias for `export_signed_bundle`."""
        return self.export_signed_bundle(*args, **kwargs)

    def ingest_signed_bundle(
        self,
        signed_bundle: Mapping[str, Any],
    ) -> tuple[bool, str, dict[str, Any] | None]:
        """Verify Ed25519 signature and AnonymityGuard schema of an inbound catalog bundle."""
        if self.subscription.sync_mode == SyncMode.OFF:
            return False, "sync_mode_off", None

        try:
            v_raw = signed_bundle["v"]
            v_bytes = v_raw if isinstance(v_raw, (bytes, bytearray)) else str(v_raw).encode("utf-8")
            seq = int(signed_bundle["seq"])
            salt = signed_bundle.get("salt", b"armiger-catalog-v2")
            if isinstance(salt, str):
                salt = salt.encode("utf-8")
            pub_raw = signed_bundle.get("k") or signed_bundle.get("pubkey_hex")
            pub_bytes = bytes.fromhex(pub_raw) if isinstance(pub_raw, str) else bytes(pub_raw)
            sig_raw = signed_bundle.get("sig") or signed_bundle.get("sig_hex")
            sig_bytes = bytes.fromhex(sig_raw) if isinstance(sig_raw, str) else bytes(sig_raw)
        except Exception as exc:
            return False, f"malformed_bundle_envelope:{exc}", None

        try:
            vk = Ed25519PublicKey.from_public_bytes(pub_bytes)
            vk.verify(sig_bytes, _bep44_canonical_sign_buffer(v=bytes(v_bytes), seq=seq, salt=bytes(salt)))
        except Exception as exc:
            return False, f"invalid_ed25519_signature:{exc}", None

        try:
            parsed = self.anonymity_guard.inspect_wire_payload(bytes(v_bytes))
        except AnonymityViolationError as exc:
            return False, f"anonymity_firewall_rejected:{exc}", None

        return True, "verified_and_ingested", parsed

    def ingest_signed_delta_bundle(
        self,
        signed_bundle: Mapping[str, Any],
    ) -> tuple[bool, str, dict[str, Any] | None]:
        """Alias for `ingest_signed_bundle`."""
        return self.ingest_signed_bundle(signed_bundle)

    async def publish_to_dht(
        self,
        swarm: Any,
        bundle: Mapping[str, Any] | None = None,
        **kwargs: Any,
    ) -> tuple[bool, str, int]:
        """Publish an anonymous signed catalog bundle to a UDP BEP-44 swarm. Blocked when `opt_in_sharing=False`."""
        if not self.opt_in_sharing:
            self._outbound_packets_blocked_opt_out += 1
            return False, "opt_in_sharing_disabled", 0

        eff_bundle = dict(bundle) if bundle is not None else self.export_signed_bundle(**kwargs)
        if eff_bundle is None:
            self._outbound_packets_blocked_opt_out += 1
            return False, "opt_in_sharing_disabled", 0

        v_bytes = eff_bundle["v"] if isinstance(eff_bundle["v"], bytes) else str(eff_bundle["v"]).encode("utf-8")
        self.anonymity_guard.inspect_wire_payload(v_bytes)

        priv_bytes = eff_bundle.get("privkey_bytes") or self.anonymity_guard.current_privkey_bytes()
        salt_bytes = eff_bundle.get("salt", b"armiger-catalog-v2")
        if isinstance(salt_bytes, str):
            salt_bytes = salt_bytes.encode("utf-8")
        responses = await swarm.put_mutable(
            privkey_bytes=priv_bytes,
            salt=salt_bytes,
            v=v_bytes,
            seq=int(eff_bundle["seq"]),
        )
        ok_count = sum(1 for r in responses if r.get(b"y") == b"r")
        return ok_count > 0, "published_anonymous_bep44", ok_count

    def close(self) -> None:
        """Close local SQLite database connection."""
        if hasattr(self, "db") and self.db is not None:
            self.db.close()

    def __del__(self) -> None:
        try:
            self.close()
        except Exception:
            pass


# ---------------------------------------------------------------------------
# Quorum Helper Factory
# ---------------------------------------------------------------------------

def create_mock_quorum_attestations(
    manifest_digest: bytes,
    distinct_asns: Sequence[int],
    timestamp: int | None = None,
) -> list[QuorumAttestation]:
    """Generate cryptographically valid Hybrid Ed25519 + WOTS+-SHAKE256 QuorumAttestations from distinct ASNs."""
    ts = int(time.time()) if timestamp is None else int(timestamp)
    attestations: list[QuorumAttestation] = []
    for asn in distinct_asns:
        priv, pub = generate_ed25519_keypair()
        sk = Ed25519PrivateKey.from_private_bytes(priv)
        node_id = hashlib.sha1(pub + f"asn{asn}".encode()).hexdigest()
        ip_digest = hashlib.sha256(f"192.0.2.{asn % 250}".encode()).hexdigest()
        temp_att = QuorumAttestation(
            node_id=node_id,
            asn=int(asn),
            ip_digest=ip_digest,
            pubkey_hex=pub.hex(),
            signature_hex="",
            timestamp=ts,
        )
        bound_msg = temp_att.signing_payload(manifest_digest)
        sig = sk.sign(bound_msg)
        pq_priv, pq_pub = WOTSPlusSHAKE256.derive_keypair_from_ed25519_priv(priv)
        pq_sig = WOTSPlusSHAKE256.sign(pq_priv, bound_msg)

        attestations.append(QuorumAttestation(
            node_id=node_id,
            asn=int(asn),
            ip_digest=ip_digest,
            pubkey_hex=pub.hex(),
            signature_hex=sig.hex(),
            timestamp=ts,
            pq_pubkey_hex=pq_pub.hex(),
            pq_signature_hex=pq_sig.hex(),
        ))
    return attestations


ArmigerDHTDBSync = ArmigerDHTDatabaseSync