Publish Armiger v3.0.0 (Rule-3 Gold mit3): weights, calibration, cards and receipts-synced docs
25cbd40 verified | """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 | |
| # --------------------------------------------------------------------------- | |
| 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() | |
| 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() | |
| 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] | |
| 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 | |
| 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 | |
| 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)) | |
| 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 | |
| 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 | |
| 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) | |
| 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") | |
| 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 | |
| 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) | |
| 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 | |
| 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 | |
| # --------------------------------------------------------------------------- | |
| 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 | |
| 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] | |
| 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] = {} | |
| 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 | |