diff --git a/README.md b/README.md index 733aa92..b9d45d4 100644 --- a/README.md +++ b/README.md @@ -36,12 +36,21 @@ Enable it on TinyWeb's `/style` page under "Forum". - Threads are auto-pruned after 30 days (configurable, or set to 0 to keep everything) - Moderation is local: block authors, mute threads, keyword filters, and gossip block lists with peers (auto-block after 3 peer reports) +### Trust circle + +Content exchange is gated by a trust graph derived from your TinyWeb subscriptions: + +- Only peers you subscribe to (with forum enabled) are trusted sources of content +- Content from trusted peers propagates transitively: you see posts from their trusted peers, and theirs, and so on — no hop limit +- New identities have no trust path until a trusted peer vouches for them or you subscribe +- Trust is seeded from TinyWeb's subscription page — each subscribed peer's forum content (and their transitive trust network) enters your view automatically. + ## Moderation All moderation is local — it controls what you see: -- **Block author** — hides all content from that identity -- **Auto-block** — when 3+ of your peers have blocked the same identity, it blocks for you too +- **Block author** — hides all content from that identity and cascades: the blocked peer's downstream trust network (peers they vouched for) is also removed from your view. If a downstream peer has an alternate trust path from another source you trust, they survive the cascade. +- **Auto-block** — when 3+ of your peers have blocked the same identity, it blocks for you too (with cascade) - **Mute thread** — hides a thread from the listing - **Keyword filters** — hides threads matching keywords - **Instance sync** — choose which peers to sync with; unsync at any time @@ -71,6 +80,7 @@ All moderation is local — it controls what you see: - Retractions are voluntary — peers can ignore them - Block gossip can be gamed (requires collusion by 3+ peers on Reticulum) - Threads prune after 30 days by default +- Discovery is through the subscription graph rather than topic-based blooms - Best-effort maintenance ## Security @@ -79,3 +89,4 @@ All moderation is local — it controls what you see: - **Retractions are voluntary** — Retracting a thread or post sends a signal to peers, but any peer can ignore it. "Retract" is a polite request, not a guaranteed delete. - **Block gossip can be gamed** — Auto-block triggers after 3 peer reports. On Reticulum this requires 3+ real instances to collude, which is impractical at mesh scale, but is not cryptographically enforced. - **No rate limiting** — Forum POST endpoints have no throttling. Low risk since the HTTP port is localhost-only by default. +- **Trust circle** — Blocking a peer removes their downstream trust network from your view. A peer with multiple independent trust paths may survive a single block. diff --git a/tinyweb_forum/__init__.py b/tinyweb_forum/__init__.py index ffdbc9c..9219d17 100644 --- a/tinyweb_forum/__init__.py +++ b/tinyweb_forum/__init__.py @@ -1,8 +1,14 @@ +import os +import sqlite3 +import threading +import time + from tinyweb_forum.db import ForumDB from tinyweb_forum.handlers import ForumHandlers from tinyweb_forum.sync import ForumSync FORUM_ENABLED_KEY = "forum_enabled" +TRUST_REFRESH_INTERVAL = 120 class ForumPlugin: @@ -16,19 +22,52 @@ class ForumPlugin: self.identity = identity self.reticulum = reticulum self._started = False + self._data_dir = data_dir + self._core_db_path = os.path.join(data_dir, "index.db") + self._trust_refresh_thread = None def is_enabled(self): return self.fdb.get_setting(FORUM_ENABLED_KEY, "0") == "1" + def _seed_trust_from_subscriptions(self): + my_hash = self.identity.hash.hex() if self.identity else "local" + if not os.path.exists(self._core_db_path): + return + try: + core = sqlite3.connect(self._core_db_path) + rows = core.execute( + "SELECT dest_hash FROM subscriptions" + ).fetchall() + core.close() + for row in rows: + self.fdb.add_trust_source(row[0], my_hash, 0) + except Exception: + pass + + def _trust_refresh_loop(self): + while self._started: + try: + self._seed_trust_from_subscriptions() + except Exception: + pass + for _ in range(TRUST_REFRESH_INTERVAL): + if not self._started: + return + time.sleep(1) + def enable(self): self.fdb.set_setting(FORUM_ENABLED_KEY, "1") if not self._started: + self._seed_trust_from_subscriptions() self.sync.start() self._started = True + self._trust_refresh_thread = threading.Thread( + target=self._trust_refresh_loop, daemon=True + ) + self._trust_refresh_thread.start() def disable(self): self.fdb.set_setting(FORUM_ENABLED_KEY, "0") - # Keep sync running so we still receive content — disable just hides UI def handle(self, method, path, query, body, cookies=None): return self.handlers.handle(method, path, query, body, cookies) diff --git a/tinyweb_forum/db.py b/tinyweb_forum/db.py index 8f396d7..e2dc3a5 100644 --- a/tinyweb_forum/db.py +++ b/tinyweb_forum/db.py @@ -4,8 +4,6 @@ import threading import time from datetime import datetime, timedelta -from tinyweb_forum.bloom import BloomFilter - FORUM_DB = "forum.db" @@ -91,20 +89,24 @@ class ForumDB: " PRIMARY KEY (content_id, content_type)" ")" ) - db.execute( - "CREATE TABLE IF NOT EXISTS peer_filters (" - " peer_hash TEXT PRIMARY KEY," - " bloom_bytes BLOB," - " bloom_size INTEGER DEFAULT 2048," - " bloom_hashes INTEGER DEFAULT 3," - " tag_count INTEGER DEFAULT 0," - " last_seen REAL NOT NULL" - ")" - ) try: db.execute("ALTER TABLE synced_instances ADD COLUMN consecutive_failures INTEGER DEFAULT 0") except Exception: pass + db.execute( + "CREATE TABLE IF NOT EXISTS forum_trust_sources (" + " instance_hash TEXT NOT NULL," + " source_hash TEXT NOT NULL," + " hops INTEGER NOT NULL," + " PRIMARY KEY (instance_hash, source_hash)" + ")" + ) + db.execute( + "CREATE TABLE IF NOT EXISTS forum_trust (" + " instance_hash TEXT PRIMARY KEY," + " min_hops INTEGER NOT NULL" + ")" + ) db.commit() db.close() @@ -611,70 +613,107 @@ class ForumDB: # --- Filter Table (Architecture B: Bloom Gossip) --- - def store_peer_filter(self, peer_hash, bloom_bytes, bloom_size, bloom_hashes, tag_count): + # --- Forum Trust --- + + # --- Forum Trust --- + + def add_trust_source(self, instance_hash, source_hash, hops): db = self.get_db() try: db.execute( - "INSERT OR REPLACE INTO peer_filters " - "(peer_hash, bloom_bytes, bloom_size, bloom_hashes, tag_count, last_seen) " - "VALUES (?, ?, ?, ?, ?, ?)", - (peer_hash, bloom_bytes, bloom_size, bloom_hashes, tag_count, time.time()), + "INSERT OR IGNORE INTO forum_trust_sources " + "(instance_hash, source_hash, hops) VALUES (?, ?, ?)", + (instance_hash, source_hash, hops), ) + existing = db.execute( + "SELECT hops FROM forum_trust_sources " + "WHERE instance_hash=? AND source_hash=?", + (instance_hash, source_hash), + ).fetchone() + if existing and hops < existing["hops"]: + db.execute( + "UPDATE forum_trust_sources SET hops=? " + "WHERE instance_hash=? AND source_hash=?", + (hops, instance_hash, source_hash), + ) + self._recompute_trust_view(db) db.commit() finally: self.return_db(db) - def get_peer_filter(self, peer_hash): + def remove_trust_source(self, source_hash): + db = self.get_db() + try: + rows = db.execute( + "WITH RECURSIVE cascade(h) AS (" + " VALUES(?)" + " UNION ALL" + " SELECT fts.instance_hash FROM forum_trust_sources fts" + " INNER JOIN cascade c ON fts.source_hash = c.h" + ") SELECT h FROM cascade", + (source_hash,), + ).fetchall() + cascade_hashes = [r["h"] for r in rows] + placeholders = ",".join("?" for _ in cascade_hashes) + db.execute( + f"DELETE FROM forum_trust_sources WHERE source_hash IN ({placeholders})", + cascade_hashes, + ) + self._recompute_trust_view(db) + db.commit() + finally: + self.return_db(db) + + def remove_by_hash(self, instance_hash): + self.remove_trust_source(instance_hash) + db = self.get_db() + try: + db.execute( + "DELETE FROM forum_trust_sources WHERE instance_hash=?", + (instance_hash,), + ) + self._recompute_trust_view(db) + db.commit() + finally: + self.return_db(db) + + def is_trusted(self, instance_hash): db = self.get_db() try: row = db.execute( - "SELECT * FROM peer_filters WHERE peer_hash = ?", (peer_hash,) + "SELECT 1 FROM forum_trust WHERE instance_hash=?", (instance_hash,) ).fetchone() - if row: - return dict(row) - return None + return row is not None finally: self.return_db(db) - def get_all_filters(self): - """Return all stored peer filters (for filter table gossip).""" + def get_trust_hops(self, instance_hash): db = self.get_db() try: - return [dict(r) for r in db.execute( - "SELECT * FROM peer_filters ORDER BY last_seen DESC" - ).fetchall()] + row = db.execute( + "SELECT min_hops FROM forum_trust WHERE instance_hash=?", (instance_hash,) + ).fetchone() + return row["min_hops"] if row else None finally: self.return_db(db) - def get_filtered_peers_by_tag(self, tag): - """Return peer hashes whose bloom filter might contain the given tag.""" - tag = tag.strip().lower() + def get_all_trusted_hashes(self): db = self.get_db() try: - matches = [] - for r in db.execute("SELECT * FROM peer_filters").fetchall(): - bf = BloomFilter.from_bytes(r["bloom_bytes"], r["bloom_size"], r["bloom_hashes"]) - if bf.might_contain(tag): - matches.append(r["peer_hash"]) - return matches + return [ + r["instance_hash"] for r in db.execute( + "SELECT instance_hash FROM forum_trust" + ).fetchall() + ] finally: self.return_db(db) - def prune_peer_filters(self, max_age_days=7): - db = self.get_db() - try: - cutoff = time.time() - max_age_days * 86400 - db.execute("DELETE FROM peer_filters WHERE last_seen < ?", (cutoff,)) - db.commit() - finally: - self.return_db(db) - - def get_peer_filter_count(self): - db = self.get_db() - try: - return db.execute("SELECT count(*) FROM peer_filters").fetchone()[0] - finally: - self.return_db(db) + def _recompute_trust_view(self, db): + db.execute("DELETE FROM forum_trust") + db.execute( + "INSERT INTO forum_trust (instance_hash, min_hops) " + "SELECT instance_hash, MIN(hops) FROM forum_trust_sources GROUP BY instance_hash" + ) # --- Peer Liveness --- diff --git a/tinyweb_forum/handlers.py b/tinyweb_forum/handlers.py index 343dd6e..cbe3097 100644 --- a/tinyweb_forum/handlers.py +++ b/tinyweb_forum/handlers.py @@ -190,12 +190,10 @@ class ForumHandlers: topics = self._get_subscribed_topics() topics_str = ", ".join(topics) if topics else "everything" peer_count = len(self.fdb.get_synced_instances()) - filter_count = self.fdb.get_peer_filter_count() return ( f'
' f'subscribed: {esc(topics_str)}' f' · {peer_count} peers' - f' · {filter_count} filters' f'
' ) @@ -721,6 +719,7 @@ class ForumHandlers: blocked = self._blocked_instances() blocked.add(instance) self.fdb.set_setting("blocked_instances", ",".join(blocked)) + self.fdb.remove_by_hash(instance) self._set_flash(f"Blocked {instance[:16]}...") return self._redirect("/forum/moderation") @@ -744,6 +743,7 @@ class ForumHandlers: self.fdb.clear_peer_block(instance) else: blocked.add(instance) + self.fdb.remove_by_hash(instance) self.fdb.set_setting("blocked_instances", ",".join(blocked)) auto = set(h.strip() for h in self.fdb.get_setting("auto_blocked_instances", "").split(",") if h.strip()) auto.discard(instance) @@ -839,75 +839,24 @@ class ForumHandlers: incoming_threads = data.get("threads", []) incoming_posts = data.get("posts", []) incoming_upvotes = data.get("upvotes", []) - peer_topics = data.get("my_topics", []) - peer_tag_cloud = data.get("my_tag_cloud", []) peer_bloom_data = data.get("content_bloom") peer_tag_bloom_data = data.get("tag_bloom") - peer_filter_table = data.get("filter_table", {}) - scoped_query_tag = data.get("scoped_query", "") blocked = self._blocked_instances() my_topics = self._get_subscribed_topics() from_hash = data.get("from_hash", "") - # Store peer's tag bloom filter (Architecture B discovery) - if peer_tag_bloom_data and from_hash: - self.fdb.store_peer_filter( - peer_hash=from_hash, - bloom_bytes=bytes(peer_tag_bloom_data), - bloom_size=data.get("tag_bloom_size", 2048), - bloom_hashes=data.get("tag_bloom_hashes", 3), - tag_count=len(peer_topics), - ) - - # Merge filter table gossip (transitive peer discovery) - if peer_filter_table and from_hash: - for ph, entry in peer_filter_table.items(): - if isinstance(entry, dict): - bb = entry.get("bloom_bytes") - if bb: - self.fdb.store_peer_filter( - peer_hash=ph, - bloom_bytes=bytes(bb) if isinstance(bb, list) else bb, - bloom_size=entry.get("bloom_size", 2048), - bloom_hashes=entry.get("bloom_hashes", 3), - tag_count=entry.get("tag_count", 0), - ) - - # Handle scoped query: find peers whose bloom might contain the queried tag - scoped_query_results = [] - if scoped_query_tag and from_hash: - scoped_query_results = self.fdb.get_filtered_peers_by_tag(scoped_query_tag) - - # Store peer's topics for future routing (backward compat) - if peer_topics and from_hash: - self.fdb.set_setting(f"peer_topics_{from_hash}", ",".join(peer_topics)) - if peer_tag_cloud and from_hash: - self.fdb.set_setting(f"peer_tag_cloud_{from_hash}", json.dumps(peer_tag_cloud)) - - # Build our tag bloom filter to send back - peer_tag_bs = data.get("tag_bloom_size", 2048) - peer_tag_bh = data.get("tag_bloom_hashes", 3) - my_tag_bloom = BloomFilter.from_tags(my_topics or [], peer_tag_bs, peer_tag_bh) + # Merge trust gossip from requester + if from_hash: + self.sync._merge_trust_gossip(data.get("trust_list", []), from_hash) # Build our content bloom filter for dedup our_existing = set() + peer_topics = data.get("my_topics", []) for t in self.fdb.get_threads_by_topics(peer_topics) if peer_topics else []: our_existing.add(t["id"]) our_bloom = BloomFilter.from_items(list(our_existing), data.get("bloom_size", 2048) if data else 2048, data.get("bloom_hashes", 3) if data else 3) - # Build filter table gossip from our stored filters - all_filters = self.fdb.get_all_filters() - filter_table_gossip = {} - for f in all_filters[:20]: - if f["peer_hash"] != from_hash: - filter_table_gossip[f["peer_hash"]] = { - "bloom_bytes": list(f["bloom_bytes"]), - "bloom_size": f["bloom_size"], - "bloom_hashes": f["bloom_hashes"], - "tag_count": f["tag_count"], - } - # Parse peer's content bloom for dedup peer_bloom = None if peer_bloom_data: @@ -920,7 +869,10 @@ class ForumHandlers: # Merge incoming content, filtered by bloom and topics if incoming_threads: for t in incoming_threads: - if t.get("author_instance", "") in blocked: + author = t.get("author_instance", "") + if author in blocked: + continue + if author and not self.fdb.is_trusted(author): continue if peer_bloom and peer_bloom.might_contain(t["id"]): continue @@ -932,7 +884,10 @@ class ForumHandlers: self.fdb.merge_thread(t) if incoming_posts: for p in incoming_posts: - if p.get("author_instance", "") in blocked: + author = p.get("author_instance", "") + if author in blocked: + continue + if author and not self.fdb.is_trusted(author): continue if peer_bloom and peer_bloom.might_contain(p["id"]): continue @@ -963,8 +918,6 @@ class ForumHandlers: my_blocks = list(blocked) my_peer_blocks = self.fdb.get_peer_block_list() - my_tag_cloud = self.fdb.get_tag_cloud(50) - my_tag_list = [t for t, _ in my_tag_cloud] threads, posts, upvote_threads = [], [], [] if since: @@ -981,12 +934,14 @@ class ForumHandlers: posts = [dict(r) for r in posts_list] upvote_threads = up_list + # Filter outgoing content by requester's trust list + requester_trust = set(data.get("trust_list", [])) + if requester_trust: + threads = [t for t in threads if t.get("author_instance", "") in requester_trust] + thread_ids = set(t["id"] for t in threads) + posts = [p for p in posts if p.get("thread_id") in thread_ids] + known_peers = [h for h in self.fdb.get_all_known_hashes() if h != from_hash] - peer_topics_map = {} - for ph in known_peers[:100]: - pt = self.fdb.get_setting(f"peer_topics_{ph}", "") - if pt: - peer_topics_map[ph] = [t.strip() for t in pt.split(",") if t.strip()] retracted = [{"id": cid, "type": ct, "author": ai, "at": ra} for cid, ct, ai, ra in self.fdb.get_raw_retractions()] @@ -1001,18 +956,10 @@ class ForumHandlers: "blocks": {"mine": my_blocks, "peers": my_peer_blocks}, "retractions": retracted, "known_peers": known_peers, - "peer_topics": my_tag_list, - "peer_tag_cloud": my_tag_cloud, "content_bloom": list(our_bloom.bytes), "bloom_size": data.get("bloom_size", 2048), "bloom_hashes": data.get("bloom_hashes", 3), - "peer_topics_map": peer_topics_map, - # Architecture B additions - "tag_bloom": list(my_tag_bloom.bytes), - "tag_bloom_size": peer_tag_bs, - "tag_bloom_hashes": peer_tag_bh, - "filter_table": filter_table_gossip, - "scoped_query_results": scoped_query_results, + "trust_list": self.fdb.get_all_trusted_hashes(), }), "headers": {}, } diff --git a/tinyweb_forum/sync.py b/tinyweb_forum/sync.py index a9d544d..e27b763 100644 --- a/tinyweb_forum/sync.py +++ b/tinyweb_forum/sync.py @@ -12,11 +12,7 @@ REQUEST_TIMEOUT = 60 GOSSIP_FANOUT = 20 BLOOM_SIZE = 2048 BLOOM_HASHES = 3 -TAG_BLOOM_SIZE = 2048 -TAG_BLOOM_HASHES = 3 -FILTER_TABLE_GOSSIP = 20 MAX_PEER_FAILURES = 5 -FILTER_TABLE_TTL_DAYS = 7 class _ForumAnnounceHandler: @@ -50,21 +46,23 @@ class ForumSync: raw = self.fdb.get_setting("topic_subscriptions", "") return [t.strip().lower() for t in raw.split(",") if t.strip()] - def _build_tag_bloom(self, topics=None): - if topics is None: - topics = self._get_subscribed_topics() - return BloomFilter.from_tags(topics or [], TAG_BLOOM_SIZE, TAG_BLOOM_HASHES) - - def _topics_overlap(self, my_topics, their_topics): - if not my_topics or not their_topics: - return True - return bool(set(my_topics) & set(their_topics)) - - def _topics_overlap_bloom(self, my_topics, peer_bloom): - """Check overlap using bloom filter instead of topic list.""" - if not my_topics or peer_bloom is None: - return True - return any(peer_bloom.might_contain(t) for t in my_topics) + def _merge_trust_gossip(self, trust_list, source_hash): + source_hops = self.fdb.get_trust_hops(source_hash) + if source_hops is None: + return + my_hash = self.identity.hash.hex() if self.identity else "local" + blocked = set( + h.strip() + for h in self.fdb.get_setting("blocked_instances", "").split(",") + if h.strip() + ) + for peer_hash in trust_list: + if peer_hash in blocked or peer_hash == my_hash: + continue + new_hops = source_hops + 1 + existing = self.fdb.get_trust_hops(peer_hash) + if existing is None or new_hops < existing: + self.fdb.add_trust_source(peer_hash, source_hash, new_hops) def start(self): self.destination = RNS.Destination( @@ -99,43 +97,17 @@ class ForumSync: random.shuffle(instances) count = 0 my_topics = self._get_subscribed_topics() - my_tag_bloom = self._build_tag_bloom(my_topics) for inst in instances[:GOSSIP_FANOUT]: if not self._running: break - pfilter = self.fdb.get_peer_filter(inst["instance_hash"]) - if pfilter: - pbloom = BloomFilter.from_bytes(pfilter["bloom_bytes"], pfilter["bloom_size"], pfilter["bloom_hashes"]) - if not self._topics_overlap_bloom(my_topics, pbloom): - continue - else: - peer_topics = self._peer_tag_topics(inst["instance_hash"]) - if not self._topics_overlap(my_topics, peer_topics): - continue try: - self._sync_with(inst["instance_hash"], my_topics, my_tag_bloom) + self._sync_with(inst["instance_hash"], my_topics) count += 1 except Exception as e: self.fdb.record_sync_result(inst["instance_hash"], False) print(f"[forum] sync error with {inst['instance_hash'][:16]}: {e}") return count - def scoped_query(self, tag): - """Ask known peers if they know anyone with the given tag.""" - tag = tag.strip().lower() - instances = self.fdb.get_synced_instances() - random.shuffle(instances) - results = set() - for inst in instances[:GOSSIP_FANOUT]: - if not self._running: - break - try: - matches = self._scoped_query_peer(inst["instance_hash"], tag) - results.update(matches) - except Exception: - continue - return list(results) - def _start_sync_loop(self): if self._thread and self._thread.is_alive(): return @@ -173,22 +145,12 @@ class ForumSync: try: instances = self.fdb.get_synced_instances() my_topics = self._get_subscribed_topics() - my_tag_bloom = self._build_tag_bloom(my_topics) random.shuffle(instances) for inst in instances[:GOSSIP_FANOUT]: if not self._running: break - pfilter = self.fdb.get_peer_filter(inst["instance_hash"]) - if pfilter: - pbloom = BloomFilter.from_bytes(pfilter["bloom_bytes"], pfilter["bloom_size"], pfilter["bloom_hashes"]) - if not self._topics_overlap_bloom(my_topics, pbloom): - continue - else: - peer_topics = self._peer_tag_topics(inst["instance_hash"]) - if not self._topics_overlap(my_topics, peer_topics): - continue try: - self._sync_with(inst["instance_hash"], my_topics, my_tag_bloom) + self._sync_with(inst["instance_hash"], my_topics) except Exception as e: self.fdb.record_sync_result(inst["instance_hash"], False) print(f"[forum] sync error with {inst['instance_hash'][:16]}: {e}") @@ -202,10 +164,6 @@ class ForumSync: self.fdb.prune_old_content(days) except Exception: pass - try: - self.fdb.prune_peer_filters(FILTER_TABLE_TTL_DAYS) - except Exception: - pass try: for dead in self.fdb.get_dead_peers(MAX_PEER_FAILURES): self.fdb.remove_synced_instance(dead) @@ -216,84 +174,7 @@ class ForumSync: return time.sleep(1) - def _peer_tag_topics(self, instance_hash): - raw = self.fdb.get_setting(f"peer_topics_{instance_hash}", "") - return [t.strip().lower() for t in raw.split(",") if t.strip()] if raw else [] - - def _store_peer_filter(self, peer_hash, data): - tag_bloom_data = data.get("tag_bloom") - if tag_bloom_data: - self.fdb.store_peer_filter( - peer_hash=peer_hash, - bloom_bytes=bytes(tag_bloom_data), - bloom_size=data.get("tag_bloom_size", TAG_BLOOM_SIZE), - bloom_hashes=data.get("tag_bloom_hashes", TAG_BLOOM_HASHES), - tag_count=len(data.get("peer_topics", [])), - ) - - def _merge_filter_table_gossip(self, filter_table): - if not filter_table: - return - for ph, entry in filter_table.items(): - if isinstance(entry, dict): - bb = entry.get("bloom_bytes") - if bb: - self.fdb.store_peer_filter( - peer_hash=ph, - bloom_bytes=bytes(bb) if isinstance(bb, list) else bb, - bloom_size=entry.get("bloom_size", TAG_BLOOM_SIZE), - bloom_hashes=entry.get("bloom_hashes", TAG_BLOOM_HASHES), - tag_count=entry.get("tag_count", 0), - ) - - def _scoped_query_peer(self, peer_hash, tag): - dest_hash = bytes.fromhex(peer_hash) - if not RNS.Transport.has_path(dest_hash): - RNS.Transport.request_path(dest_hash) - elapsed = 0 - while not RNS.Transport.has_path(dest_hash) and elapsed < 15: - time.sleep(0.5) - elapsed += 0.5 - if not RNS.Transport.has_path(dest_hash): - return [] - server_identity = RNS.Identity.recall(dest_hash) - if server_identity is None: - return [] - destination = RNS.Destination( - server_identity, - RNS.Destination.OUT, - RNS.Destination.SINGLE, - FORUM_APP, - ) - link = RNS.Link(destination) - elapsed = 0 - while link.status == RNS.Link.PENDING and elapsed < 15: - time.sleep(0.25) - elapsed += 0.25 - if link.status != RNS.Link.ACTIVE: - return [] - try: - my_hash = self.identity.hash.hex() if self.identity else "local" - request_data = { - "scoped_query": tag, - "from_hash": my_hash, - } - receipt = link.request("/forum", data=request_data, timeout=REQUEST_TIMEOUT) - elapsed = 0 - done = (RNS.RequestReceipt.READY, RNS.RequestReceipt.DELIVERED, RNS.RequestReceipt.FAILED) - while receipt.get_status() not in done and elapsed < REQUEST_TIMEOUT: - time.sleep(0.1) - elapsed += 0.1 - if receipt.get_status() in (RNS.RequestReceipt.READY, RNS.RequestReceipt.DELIVERED): - resp = receipt.get_response() - if isinstance(resp, dict) and resp.get("status") == 200: - data = json.loads(resp["body"]) - return data.get("scoped_query_results", []) - return [] - finally: - link.teardown() - - def _sync_with(self, instance_hash, my_topics=None, my_tag_bloom=None): + def _sync_with(self, instance_hash, my_topics=None): success = False dest_hash = bytes.fromhex(instance_hash) if not RNS.Transport.has_path(dest_hash): @@ -341,25 +222,8 @@ class ForumSync: if my_topics is None: my_topics = self._get_subscribed_topics() - if my_tag_bloom is None: - my_tag_bloom = self._build_tag_bloom(my_topics) - # Build filter table gossip: share a subset of our known peer filters - all_filters = self.fdb.get_all_filters() - filter_table_gossip = {} - for f in all_filters[:FILTER_TABLE_GOSSIP]: - if f["peer_hash"] != instance_hash and f["peer_hash"] != my_hash: - filter_table_gossip[f["peer_hash"]] = { - "bloom_bytes": list(f["bloom_bytes"]), - "bloom_size": f["bloom_size"], - "bloom_hashes": f["bloom_hashes"], - "tag_count": f["tag_count"], - } - - threads, posts = [], [] - upvotes = [] - my_tag_cloud = self.fdb.get_tag_cloud(50) - my_tag_list = [t for t, _ in my_tag_cloud] + threads, posts, upvotes = [], [], [] # Only build content if we have a since timestamp (incremental sync) if since: @@ -389,11 +253,6 @@ class ForumSync: for cid, ct, ai, ra in self.fdb.get_raw_retractions()] known_peers = [h for h in self.fdb.get_all_known_hashes() if h != instance_hash and h != my_hash] - peer_topics_map = {} - for ph in known_peers[:100]: - pt = self.fdb.get_setting(f"peer_topics_{ph}", "") - if pt: - peer_topics_map[ph] = [t.strip() for t in pt.split(",") if t.strip()] request_data = { "query": {"since": [since]} if since else {}, @@ -405,16 +264,10 @@ class ForumSync: "retractions": retracted, "known_peers": known_peers, "my_topics": my_topics, - "my_tag_cloud": my_tag_cloud, "content_bloom": list(content_bloom.bytes), "bloom_size": BLOOM_SIZE, "bloom_hashes": BLOOM_HASHES, - "peer_topics_map": peer_topics_map, - # Architecture B additions - "tag_bloom": list(my_tag_bloom.bytes), - "tag_bloom_size": TAG_BLOOM_SIZE, - "tag_bloom_hashes": TAG_BLOOM_HASHES, - "filter_table": filter_table_gossip, + "trust_list": self.fdb.get_all_trusted_hashes(), } receipt = link.request("/forum", data=request_data, timeout=REQUEST_TIMEOUT) @@ -432,52 +285,10 @@ class ForumSync: except (json.JSONDecodeError, KeyError): data = {} success = True - - # Store peer's tag bloom filter - self._store_peer_filter(instance_hash, data) - - # Merge filter table gossip from peer - self._merge_filter_table_gossip(data.get("filter_table", {})) - - # Store peer topics (backward compat) - peer_topics = data.get("peer_topics", []) - if peer_topics: - self.fdb.set_setting(f"peer_topics_{instance_hash}", ",".join(peer_topics)) - peer_tag_cloud = data.get("peer_tag_cloud", []) - if peer_tag_cloud: - self.fdb.set_setting(f"peer_tag_cloud_{instance_hash}", json.dumps(peer_tag_cloud)) - - # Check tag overlap using bloom filter my_blocks = set(h.strip() for h in self.fdb.get_setting("blocked_instances", "").split(",") if h.strip()) - peer_tag_bloom_data = data.get("tag_bloom") - peer_tag_bloom = None - if peer_tag_bloom_data: - peer_tag_bloom = BloomFilter.from_bytes( - bytes(peer_tag_bloom_data), - data.get("tag_bloom_size", TAG_BLOOM_SIZE), - data.get("tag_bloom_hashes", TAG_BLOOM_HASHES), - ) - - # If no tag overlap, skip content sync but still store the filter - if my_topics and peer_tag_bloom is not None: - if not self._topics_overlap_bloom(my_topics, peer_tag_bloom): - success = True - now = time.strftime("%Y-%m-%dT%H:%M:%S") - self.fdb.set_last_sync(instance_hash, now) - self.fdb.record_sync_result(instance_hash, True) - return - - # Fall back to topic list overlap check if no bloom - if peer_tag_bloom is None: - incoming_topics = data.get("peer_topics", []) - if my_topics and incoming_topics: - if not self._topics_overlap(my_topics, incoming_topics): - success = True - now = time.strftime("%Y-%m-%dT%H:%M:%S") - self.fdb.set_last_sync(instance_hash, now) - self.fdb.record_sync_result(instance_hash, True) - return + # Merge trust gossip from peer + self._merge_trust_gossip(data.get("trust_list", []), instance_hash) # Check content bloom for dedup peer_bloom_data = data.get("content_bloom") @@ -490,20 +301,28 @@ class ForumSync: ) for t in data.get("threads", []): - if t.get("author_instance", "") not in my_blocks: - if peer_bloom and peer_bloom.might_contain(t["id"]): - continue - if not my_topics: + author = t.get("author_instance", "") + if author in my_blocks: + continue + if author and not self.fdb.is_trusted(author): + continue + if peer_bloom and peer_bloom.might_contain(t["id"]): + continue + if not my_topics: + self.fdb.merge_thread(t) + else: + t_tags = [tag.strip().lower() for tag in t.get("tags", "").split(",") if tag.strip()] + if set(my_topics) & set(t_tags): self.fdb.merge_thread(t) - else: - t_tags = [tag.strip().lower() for tag in t.get("tags", "").split(",") if tag.strip()] - if set(my_topics) & set(t_tags): - self.fdb.merge_thread(t) for p in data.get("posts", []): - if p.get("author_instance", "") not in my_blocks: - if peer_bloom and peer_bloom.might_contain(p["id"]): - continue - self.fdb.merge_post(p) + author = p.get("author_instance", "") + if author in my_blocks: + continue + if author and not self.fdb.is_trusted(author): + continue + if peer_bloom and peer_bloom.might_contain(p["id"]): + continue + self.fdb.merge_post(p) for tid in data.get("upvote_threads", []): self.fdb.merge_upvote(tid, instance_hash) @@ -524,13 +343,6 @@ class ForumSync: if peer_hash and peer_hash != my_hash and peer_hash != instance_hash: self.fdb.add_known_peer(peer_hash) - my_topics_set = set(my_topics) - for ph, pt in data.get("peer_topics_map", {}).items(): - if ph and ph != my_hash and ph != instance_hash: - pt_set = set(t.lower() for t in pt) - if not my_topics or my_topics_set & pt_set: - self.fdb.add_known_peer(ph) - now = time.strftime("%Y-%m-%dT%H:%M:%S") self.fdb.set_last_sync(instance_hash, now) self.fdb.record_sync_result(instance_hash, success) @@ -546,6 +358,7 @@ class ForumSync: if h not in blocked and h not in auto_blocked and count >= 3: blocked.add(h) auto_blocked.add(h) + self.fdb.remove_by_hash(h) changed = True if changed: self.fdb.set_setting("blocked_instances", ",".join(blocked))