"""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 io 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" SQLITE_MAX_INT64 = (1 << 63) - 1 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 (Deterministic Mandatory Zlib Wire Codec + Optional Zstd) # --------------------------------------------------------------------------- MANDATORY_WIRE_CODEC = "zlib" ZSTD_MAGIC = b"\x28\xb5\x2f\xfd" MAX_DECOMPRESSED_SHARD_BYTES = 4 * 1024 * 1024 MAX_DECOMPRESSION_RATIO = 50 def compress_payload(data: bytes, codec: str = MANDATORY_WIRE_CODEC) -> tuple[bytes, str]: """Compress payload using the mandatory deterministic wire codec ('zlib' level 9) by default. All peers in the swarm MUST produce identical compressed bytes for the same record set so that content-addressed shard digests and BEP-46 infohashes match across hosts regardless of whether optional `zstandard` is installed. Explicit `codec='zstd'` is supported only when `zstandard` is available. """ norm_codec = str(codec or MANDATORY_WIRE_CODEC).strip().lower() if norm_codec == "zlib": return zlib.compress(data, level=9), "zlib" if norm_codec == "zstd": if not _HAS_ZSTD: raise RuntimeError("zstd_codec_unavailable: optional zstandard package is not installed") cctx = zstd.ZstdCompressor(level=7) return cctx.compress(data), "zstd" raise ValueError(f"Unsupported compression codec: {codec}") def decompress_payload( compressed: bytes, codec: str = MANDATORY_WIRE_CODEC, max_output_bytes: int = MAX_DECOMPRESSED_SHARD_BYTES, max_ratio: int = MAX_DECOMPRESSION_RATIO, *, max_uncompressed_bytes: int | None = None, ) -> bytes: """Decompress payload using specified codec with strict zip-bomb size & ratio caps.""" if not compressed: return b"" if max_uncompressed_bytes is not None: max_output_bytes = int(max_uncompressed_bytes) hard_cap = min(max_output_bytes, max(65536, len(compressed) * max_ratio)) def _zlib_bounded(data: bytes) -> bytes: try: dobj = zlib.decompressobj() out = dobj.decompress(data, hard_cap + 1) if dobj.unconsumed_tail or len(out) > hard_cap: return b"" out += dobj.flush() if len(out) > hard_cap: return b"" return out except Exception: return b"" norm_codec = str(codec or MANDATORY_WIRE_CODEC).strip().lower() is_zstd_frame = compressed.startswith(ZSTD_MAGIC) if norm_codec == "zlib" and not is_zstd_frame: return _zlib_bounded(compressed) if norm_codec in ("zstd", "zlib") and is_zstd_frame: if not _HAS_ZSTD: return b"" try: params = zstd.get_frame_parameters(compressed) if ( params.content_size not in (0, zstd.CONTENTSIZE_UNKNOWN) and params.content_size > hard_cap ): return b"" dctx = zstd.ZstdDecompressor() with dctx.stream_reader(io.BytesIO(compressed)) as reader: out = reader.read(hard_cap + 1) if len(out) > hard_cap: return b"" return out except Exception: return b"" if norm_codec == "zstd": 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, timeout=10.0) 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, safety_gate_passed INTEGER NOT NULL DEFAULT 1, created_at INTEGER NOT NULL ) """) try: self.conn.execute("ALTER TABLE mainstream_torrents ADD COLUMN safety_gate_passed INTEGER NOT NULL DEFAULT 1") except sqlite3.OperationalError: pass 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)") self.conn.execute(""" CREATE TABLE IF NOT EXISTS shard_sequences ( shard_id TEXT PRIMARY KEY, seq INTEGER NOT NULL, updated_at INTEGER NOT NULL ) """) # 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, safety_gate_passed INTEGER NOT NULL DEFAULT 1, created_at INTEGER NOT NULL ) """) try: self.conn.execute("ALTER TABLE adult_torrents ADD COLUMN safety_gate_passed INTEGER NOT NULL DEFAULT 1") except sqlite3.OperationalError: pass self.conn.execute("CREATE INDEX IF NOT EXISTS idx_adult_lang ON adult_torrents(language)") def get_shard_sequences(self) -> dict[str, int]: """Return durable per-shard monotonic sequence high-water marks from SQLite.""" cur = self.conn.cursor() cur.execute("SELECT shard_id, seq FROM shard_sequences") return {str(row["shard_id"]): int(row["seq"]) for row in cur.fetchall()} def insert_record(self, record: TorrentRecord) -> bool: """Insert a torrent record into the appropriate partitioned table.""" if not record.safety_gate_passed: return False for fname, fval in ( ("size_bytes", record.size_bytes), ("file_count", record.file_count), ("seeders", record.seeders), ("leechers", record.leechers), ("created_at", record.created_at), ): if isinstance(fval, bool) or not isinstance(fval, int) or fval < 0: raise ValueError(f"invalid_shard_record_int_field:{fname}") if fval > SQLITE_MAX_INT64: raise ValueError(f"integer_overflow_for_sqlite_int64:{fname}") 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, safety_gate_passed, 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), 1 if record.safety_gate_passed else 0, record.created_at, ), ) return True def insert_records_batch( self, records: Sequence[TorrentRecord], shard_id: str | None = None, seq: int | None = None, ) -> int: """Batch-insert validated torrent records and advance shard sequence within a single SQLite transaction.""" if not records and (shard_id is None or seq is None): return 0 main_rows: list[tuple[Any, ...]] = [] adult_rows: list[tuple[Any, ...]] = [] for record in records: if not record.safety_gate_passed: raise ValueError(f"record_failed_safety_gate:{record.infohash}") for fname, fval in ( ("size_bytes", record.size_bytes), ("file_count", record.file_count), ("seeders", record.seeders), ("leechers", record.leechers), ("created_at", record.created_at), ): if isinstance(fval, bool) or not isinstance(fval, int) or fval < 0: raise ValueError(f"invalid_shard_record_int_field:{fname}") if fval > SQLITE_MAX_INT64: raise ValueError(f"integer_overflow_for_sqlite_int64:{fname}") 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), 1 if record.safety_gate_passed else 0, record.created_at, ) if is_adult: adult_rows.append(row) else: main_rows.append(row) if self.conn.in_transaction: self.conn.commit() self.conn.execute("BEGIN IMMEDIATE") try: if shard_id is not None and seq is not None: norm_shard = shard_id.strip().lower() cur_row = self.conn.execute( "SELECT seq FROM shard_sequences WHERE shard_id = ?", (norm_shard,), ).fetchone() if cur_row is not None and int(seq) <= int(cur_row["seq"]): raise ValueError(f"stale_or_downgrade_sequence:{seq}<={cur_row['seq']}") 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, safety_gate_passed, 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, safety_gate_passed, created_at ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, adult_rows, ) if shard_id is not None and seq is not None: self.conn.execute( """ INSERT INTO shard_sequences (shard_id, seq, updated_at) VALUES (?, ?, ?) ON CONFLICT(shard_id) DO UPDATE SET seq = excluded.seq, updated_at = excluded.updated_at WHERE excluded.seq > shard_sequences.seq """, (shard_id.strip().lower(), int(seq), int(time.time())), ) self.conn.commit() except Exception: if self.conn.in_transaction: self.conn.rollback() raise 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=bool(row["safety_gate_passed"]), 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=bool(row["safety_gate_passed"]), 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], *, max_output_bytes: int = MAX_DECOMPRESSED_SHARD_BYTES, max_ratio: int = MAX_DECOMPRESSION_RATIO, ) -> tuple[bytes, MerkleTree]: """Serialize a list of TorrentRecords into a deterministic compressed micro-bundle. Enforces `MAX_DECOMPRESSED_SHARD_BYTES` and `MAX_DECOMPRESSION_RATIO` at build time before any one-time WOTS+ signature leaf is reserved or consumed. """ # 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") if len(raw_json) > max_output_bytes: raise ValueError( f"shard_exceeds_decompression_bounds:uncompressed_bytes={len(raw_json)}>{max_output_bytes}" ) compressed, _codec = compress_payload(raw_json, codec=MANDATORY_WIRE_CODEC) hard_cap = min(max_output_bytes, max(65536, len(compressed) * max_ratio)) if len(raw_json) > hard_cap: raise ValueError( f"shard_exceeds_decompression_bounds:ratio_cap_exceeded:{len(raw_json)}>{hard_cap}" ) return compressed, tree def split_records_into_bounded_shards( records: Sequence[TorrentRecord], *, max_output_bytes: int = MAX_DECOMPRESSED_SHARD_BYTES, max_ratio: int = MAX_DECOMPRESSION_RATIO, ) -> list[list[TorrentRecord]]: """Partition a sequence of TorrentRecords into bounded micro-shards that each satisfy receiver decompression caps.""" sorted_records = sorted(records, key=lambda r: r.infohash.lower()) if not sorted_records: return [[]] batches: list[list[TorrentRecord]] = [] current_batch: list[TorrentRecord] = [] for rec in sorted_records: candidate = current_batch + [rec] try: serialize_shard_records( candidate, max_output_bytes=max_output_bytes, max_ratio=max_ratio, ) current_batch = candidate except ValueError: if not current_batch: raise batches.append(current_batch) serialize_shard_records( [rec], max_output_bytes=max_output_bytes, max_ratio=max_ratio, ) current_batch = [rec] if current_batch: batches.append(current_batch) return batches def deserialize_shard_records(compressed: bytes) -> list[TorrentRecord]: """Decompress and parse shard records with strict schema and type validation.""" raw_json = decompress_payload(compressed) payload_dict = json.loads(raw_json.decode("utf-8")) if not isinstance(payload_dict, dict): raise ValueError("shard_payload_not_dict") raw_records = payload_dict.get("records") if not isinstance(raw_records, list): raise ValueError("shard_records_not_list") records: list[TorrentRecord] = [] for d in raw_records: if not isinstance(d, dict): raise ValueError("shard_record_item_not_dict") for str_field in ("infohash", "title", "category", "language", "quality_tier", "epoch", "content_sha256"): val = d.get(str_field) if not isinstance(val, str) or not val.strip(): raise ValueError(f"invalid_shard_record_string_field:{str_field}") infohash_norm = d["infohash"].strip().lower() if len(infohash_norm) != 40 or any(ch not in "0123456789abcdef" for ch in infohash_norm): raise ValueError("invalid_shard_record_infohash_hex") sha_norm = d["content_sha256"].strip().lower() if len(sha_norm) != 64 or any(ch not in "0123456789abcdef" for ch in sha_norm): raise ValueError("invalid_shard_record_content_sha256_hex") for int_field in ("size_bytes", "file_count", "seeders", "leechers"): ival = d.get(int_field) if isinstance(ival, bool) or not isinstance(ival, int) or ival < 0: raise ValueError(f"invalid_shard_record_int_field:{int_field}") if ival > SQLITE_MAX_INT64: raise ValueError(f"integer_overflow_for_sqlite_int64:{int_field}") created_val = d.get("created_at", int(time.time())) if isinstance(created_val, bool) or not isinstance(created_val, int) or created_val < 0: raise ValueError("invalid_shard_record_int_field:created_at") if created_val > SQLITE_MAX_INT64: raise ValueError("integer_overflow_for_sqlite_int64:created_at") flist = d.get("file_list", []) if not isinstance(flist, list) or any(not isinstance(item, str) for item in flist): raise ValueError("invalid_shard_record_file_list") sg_val = d.get("safety_gate_passed", True) if not isinstance(sg_val, bool): raise ValueError("invalid_shard_record_safety_gate_passed") 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=list(flist), safety_gate_passed=sg_val, created_at=created_val, )) return records # --------------------------------------------------------------------------- # Quorum & Attestation Verification Engine # --------------------------------------------------------------------------- class QuorumVerifier: """Validates K-of-N distinct-ASN quorum attestations and Hybrid Ed25519 + WOTS+-SHAKE256 root signatures.""" _GLOBAL_ATTESTOR_ASN_REGISTRY: dict[str, int] = {} @classmethod def register_trusted_attestor(cls, pubkey_hex: str, asn: int) -> None: """Bind an attestor Ed25519 public key to its authenticated Autonomous System Number.""" norm_pk = str(pubkey_hex).strip().lower() if len(norm_pk) == 64: cls._GLOBAL_ATTESTOR_ASN_REGISTRY[norm_pk] = int(asn) 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, *, require_pq: bool = True, trusted_attestor_asns: Mapping[str, int] | None = None, ) -> None: self.trusted_roots: list[bytes] = [] self.trusted_pq_roots: list[bytes] = [ bytes.fromhex(k.strip().lower()) for k in (trusted_root_pq_pubkeys_hex or []) if len(k.strip()) == 64 ] self.ed_to_pq_roots: dict[bytes, list[bytes]] = {} raw_ed_entries = list(trusted_root_pubkeys_hex or []) for idx, raw_entry in enumerate(raw_ed_entries): cleaned = str(raw_entry).strip().lower() if ":" in cleaned: ed_part, pq_part = cleaned.split(":", 1) ed_b: bytes | None = None if len(ed_part.strip()) == 64: ed_b = bytes.fromhex(ed_part.strip()) if ed_b not in self.trusted_roots: self.trusted_roots.append(ed_b) if len(pq_part.strip()) == 64: pq_b = bytes.fromhex(pq_part.strip()) if pq_b not in self.trusted_pq_roots: self.trusted_pq_roots.append(pq_b) if ed_b is not None: bound_list = self.ed_to_pq_roots.setdefault(ed_b, []) if pq_b not in bound_list: bound_list.append(pq_b) elif len(cleaned) == 64: ed_b = bytes.fromhex(cleaned) if ed_b not in self.trusted_roots: self.trusted_roots.append(ed_b) bound_list = self.ed_to_pq_roots.setdefault(ed_b, []) if self.trusted_pq_roots: if len(self.trusted_pq_roots) == len(raw_ed_entries) and idx < len(self.trusted_pq_roots): pq_b = self.trusted_pq_roots[idx] if pq_b not in bound_list: bound_list.append(pq_b) else: bound_root_pq = WOTSPlusSHAKE256.get_root_wots_pub(cleaned) if bound_root_pq and len(bound_root_pq) == 64: pq_b = bytes.fromhex(bound_root_pq) if pq_b in self.trusted_pq_roots and pq_b not in bound_list: bound_list.append(pq_b) elif len(raw_ed_entries) == 1: for pq_b in self.trusted_pq_roots: if pq_b not in bound_list: bound_list.append(pq_b) else: bound_root_pq = WOTSPlusSHAKE256.get_root_wots_pub(cleaned) if bound_root_pq and len(bound_root_pq) == 64: pq_b = bytes.fromhex(bound_root_pq) if pq_b not in self.trusted_pq_roots: self.trusted_pq_roots.append(pq_b) if pq_b not in bound_list: bound_list.append(pq_b) self.min_quorum = min_quorum self.min_distinct_asns = min_distinct_asns self.require_pq = bool(require_pq) self.trusted_attestor_asns: dict[str, int] = { str(k).strip().lower(): int(v) for k, v in (trusted_attestor_asns or {}).items() if len(str(k).strip()) == 64 } self._used_pq_pubkeys: dict[str, bytes] = {} 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 (with authenticated ASN bindings) 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 & timestamp freshness check if manifest.seq <= current_seq: return False, f"stale_or_downgrade_sequence:{manifest.seq}<={current_seq}" if manifest.seq > (2**63 - 1) or (current_seq >= 0 and manifest.seq > current_seq + 1024): return False, f"sequence_jump_exceeds_max_window:{manifest.seq}" now_manifest_ts = int(time.time()) if ( isinstance(manifest.timestamp, bool) or not isinstance(manifest.timestamp, int) or manifest.timestamp <= 0 or (manifest.timestamp - now_manifest_ts) > 300 ): return False, f"manifest_timestamp_out_of_bounds:{manifest.timestamp}" # 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 (mandatory when require_pq=True) has_root_pq_fields = bool(manifest.pq_signature_hex or manifest.root_pq_pubkey_hex) if self.require_pq or has_root_pq_fields: if not manifest.root_pq_pubkey_hex.strip() or not manifest.pq_signature_hex.strip(): return False, "missing_root_pq_wots_signature" norm_root_pq = manifest.root_pq_pubkey_hex.strip().lower() try: pq_root_bytes = bytes.fromhex(norm_root_pq) pq_sig_bytes = bytes.fromhex(manifest.pq_signature_hex.strip()) except ValueError as exc: return False, f"invalid_root_pq_wots_signature:{exc}" bound_pq_roots = list(self.ed_to_pq_roots.get(root_pub_bytes, [])) if not bound_pq_roots: dyn_pq_hex = WOTSPlusSHAKE256.get_root_wots_pub(root_pub_bytes.hex()) if dyn_pq_hex and len(dyn_pq_hex) == 64: dyn_b = bytes.fromhex(dyn_pq_hex) if not self.trusted_pq_roots or dyn_b in self.trusted_pq_roots: bound_pq_roots.append(dyn_b) ok_pq, pq_reason = WOTSPlusSHAKE256.verify_against_trusted_roots( bound_pq_roots, pq_root_bytes, payload, pq_sig_bytes, used_keys=self._used_pq_pubkeys, ) if not ok_pq: if pq_reason == "untrusted_pq_root_pubkey": return False, f"untrusted_root_pq_pubkey:{manifest.root_pq_pubkey_hex}" if pq_reason == "wots_one_time_key_reused": return False, f"wots_one_time_key_reused:{norm_root_pq[:16]}" 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 verified attestations and distinct ASNs manifest_digest = manifest.canonical_digest() verified_attestations_count = 0 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 self.require_pq and (not att.pq_pubkey_hex.strip() or not att.pq_signature_hex.strip()): continue 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 # Verify authenticated attestor ASN binding expected_asn = ( self.trusted_attestor_asns[norm_pk] if norm_pk in self.trusted_attestor_asns else self._GLOBAL_ATTESTOR_ASN_REGISTRY.get(norm_pk) ) if expected_asn is None or int(att.asn) != int(expected_asn): continue bound_digest = att.signing_payload(manifest_digest) try: node_vk = Ed25519PublicKey.from_public_bytes(bytes.fromhex(norm_pk)) node_vk.verify(bytes.fromhex(norm_sig), bound_digest) except (InvalidSignature, ValueError): continue if self.require_pq or att.pq_signature_hex or att.pq_pubkey_hex: try: norm_att_pq = att.pq_pubkey_hex.strip().lower() node_pq_pk = bytes.fromhex(norm_att_pq) node_pq_sig = bytes.fromhex(att.pq_signature_hex.strip()) meta_att = WOTSPlusSHAKE256.extract_signature_metadata(node_pq_pk, node_pq_sig) if meta_att is None: continue att_root_pq_hex = meta_att["root_pq_pub_hex"] if not ( WOTSPlusSHAKE256.is_bound_to_ed25519_pub(norm_pk, att_root_pq_hex) or WOTSPlusSHAKE256.is_bound_to_ed25519_pub(norm_pk, norm_att_pq) ): continue prev_att_msg = self._used_pq_pubkeys.get(meta_att["one_time_key_id"]) if prev_att_msg is not None and prev_att_msg != bound_digest: continue if not WOTSPlusSHAKE256.verify(node_pq_pk, bound_digest, node_pq_sig): continue self._used_pq_pubkeys[meta_att["one_time_key_id"]] = bound_digest self._used_pq_pubkeys[norm_att_pq] = bound_digest except ValueError: continue verified_attestations_count += 1 verified_asns.add(int(att.asn)) if verified_attestations_count < self.min_quorum: return False, f"insufficient_verified_attestations:{verified_attestations_count}<{self.min_quorum}" 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, *, trusted_root_pq_pubkeys_hex: Sequence[str] | 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, trusted_root_pq_pubkeys_hex=trusted_root_pq_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, trusted_root_pq_pubkeys_hex=trusted_root_pq_pubkeys_hex, ) self.db = ShardDatabase(db_path=db_path, enable_xxx=self.subscription.enable_xxx) self.shard_sequences: dict[str, int] = self.db.get_shard_sequences() self.anonymity_guard = AnonymityGuard( ephemeral_identity=ephemeral_identity, min_k_samples=min_k_samples, catalog_secret=self.master_secret, ) 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 self._catalog_sender_seqs: dict[tuple[str, str], int] = {} 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 Merkle-authenticated leaf key sk = Ed25519PrivateKey.from_private_bytes(root_privkey) pk = sk.public_key() root_pub_bytes = pk.public_bytes(Encoding.Raw, PublicFormat.Raw) leaf_idx, pq_pub, _ = WOTSPlusSHAKE256.reserve_next_leaf(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_reserved_leaf(root_privkey, leaf_idx, 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) try: shard_cat, shard_lang, shard_qual, shard_epoch = SubscriptionConfig.parse_shard_id(manifest.shard_id) except ValueError: return False, f"malformed_shard_id:{manifest.shard_id}", 0 if shard_epoch not in ("*", "any") and manifest.epoch.strip().lower() != shard_epoch: return False, f"manifest_epoch_mismatch:{manifest.epoch}!={shard_epoch}", 0 if not self.subscription.is_subscribed(manifest.shard_id): return False, f"unsubscribed_shard_filtered:{manifest.shard_id}", 0 # 2. Verify monotonic sequence & quorum norm_shard_id = manifest.shard_id.strip().lower() db_seq = self.db.get_shard_sequences().get(norm_shard_id, -1) cur_seq = max(self.shard_sequences.get(manifest.shard_id, -1), self.shard_sequences.get(norm_shard_id, -1), db_seq) 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) 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) except Exception as exc: return False, f"shard_deserialization_failed:{exc}", 0 if reconstructed_tree.root_hex.lower() != manifest.merkle_root.lower(): return False, "merkle_root_mismatch", 0 # 6. Pre-ingest scope and safety re-inspection of decrypted records accepted_records: list[TorrentRecord] = [] for rec in records: rec_cat = rec.category.strip().lower() rec_lang = rec.language.strip().lower() rec_qual = rec.quality_tier.strip().lower() rec_ep = rec.epoch.strip().lower() if ( (shard_cat not in ("*", "any") and rec_cat != shard_cat) or (shard_lang not in ("*", "any") and rec_lang != shard_lang) or (shard_qual not in ("*", "any") and rec_qual not in (shard_qual, "any")) or (shard_epoch not in ("*", "any") and rec_ep != shard_epoch) ): return False, f"record_shard_scope_mismatch:{rec.infohash}", 0 eff_qual = shard_qual if rec_qual == "any" else rec_qual if not self.subscription.is_subscribed( SubscriptionConfig.format_shard_id(rec_cat, rec_lang, eff_qual, rec_ep) ): return False, f"record_subscription_mismatch:{rec.infohash}", 0 # Enforce safety gate passed flag and local conformal safety gate evaluation if not rec.safety_gate_passed: return False, f"record_failed_safety_gate:{rec.infohash}", 0 gate_ok, gate_reason, _ = self.safety_gate.evaluate( p_adult=1.0 if rec_cat == CATEGORY_ADULT else 0.0, p_junk=0.0, p_excluded_scope=0.0, target_category=rec_cat, ) if not gate_ok: return False, f"record_failed_safety_gate:{rec.infohash}:{gate_reason}", 0 # 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) try: ingested_count = self.db.insert_records_batch( accepted_records, shard_id=manifest.shard_id, seq=manifest.seq, ) except (ValueError, OverflowError, sqlite3.Error) as exc: return False, f"db_commit_failed:{exc}", 0 # 7. Update in-memory sequence cache after durable SQLite commit self.shard_sequences[manifest.shard_id] = manifest.seq self.shard_sequences[norm_shard_id] = manifest.seq return True, "shard_applied_successfully", ingested_count def get_shard_seq(self, shard_id: str) -> int: """Return the highest committed BEP-44 sequence number for `shard_id` (0 if unseen).""" norm_id = "/".join(SubscriptionConfig.parse_shard_id(shard_id)) return max( int(self.shard_sequences.get(shard_id, 0)), int(self.shard_sequences.get(norm_id, 0)), ) 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 keyed, domain-separated BLAKE2b-64(key, domain='armiger-catalog-v2' || normalized_key).""" return self.anonymity_guard._hash_catalog_key_with_derived_key( normalized_key, self.anonymity_guard.catalog_secret, reject_raw_pii=reject_raw_pii, ) def _outbound_allowed(self) -> tuple[bool, str]: """Return whether outbound telemetry/catalog export and DHT publishing are permitted.""" if not self.opt_in_sharing: return False, "opt_in_sharing_disabled" if self.subscription.sync_mode not in (SyncMode.SUBSCRIBE_AND_SEED, SyncMode.PUBLISHER): return False, f"sync_mode_disallows_outbound:{self.subscription.sync_mode.value}" return True, "allowed" 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` or `sync_mode` is `OFF`/`PULL_ONLY`. Raises `AnonymityViolationError` if any PII/canary/schema/k-anonymity invariant is violated. """ allowed, _ = self._outbound_allowed() if not allowed: 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, monotonic sequence high-water mark, 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") raw_seq = signed_bundle["seq"] if isinstance(raw_seq, bool) or not isinstance(raw_seq, int) or raw_seq < 1: raise ValueError(f"invalid_bundle_seq:{raw_seq!r}") seq = int(raw_seq) salt = signed_bundle.get("salt", b"armiger-catalog-v2") if isinstance(salt, str): salt = salt.encode("utf-8") salt_bytes = bytes(salt) 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 sender_key = (pub_bytes.hex().lower(), salt_bytes.hex().lower()) prev_seq = self._catalog_sender_seqs.get(sender_key, 0) if seq <= prev_seq: return False, f"stale_or_replayed_seq:{seq}<={prev_seq}", None try: vk = Ed25519PublicKey.from_public_bytes(pub_bytes) vk.verify(sig_bytes, _bep44_canonical_sign_buffer(v=bytes(v_bytes), seq=seq, salt=salt_bytes)) 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 if "seq" in parsed and int(parsed["seq"]) != seq: return False, f"envelope_wire_seq_mismatch:{seq}!={parsed.get('seq')}", None self._catalog_sender_seqs[sender_key] = seq 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` or `sync_mode` is `OFF`/`PULL_ONLY`.""" allowed, disallow_reason = self._outbound_allowed() if not allowed: self._outbound_packets_blocked_opt_out += 1 return False, disallow_reason, 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, disallow_reason, 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 idx, asn in enumerate(distinct_asns): priv, pub = generate_ed25519_keypair() QuorumVerifier.register_trusted_attestor(pub.hex(), int(asn)) sk = Ed25519PrivateKey.from_private_bytes(priv) node_id = hashlib.sha1(pub + f"asn{asn}:{idx}".encode()).hexdigest() ip_digest = hashlib.sha256(f"192.0.2.{(asn + idx) % 250}:{idx}".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