Click any node in the graph above to inspect it.
';
+ }
+
+ function countEdges(fp) {
+ let vouchOut = 0, vouchIn = 0, signalsIn = 0, knows = 0;
+ for (const e of edges) {
+ if (e.kind === "vouch" && e.source === fp) vouchOut++;
+ else if (e.kind === "vouch" && e.target === fp) vouchIn++;
+ else if (e.kind === "signal" && e.target === fp) signalsIn += e.weight;
+ else if (e.kind === "signal" && e.source === fp) signalsIn += e.weight;
+ else if (e.kind === "knows" && (e.source === fp || e.target === fp)) knows++;
+ }
+ return { vouchOut, vouchIn, signalsIn, knows };
+ }
+
+ function renderDetail(n) {
+ if (!detailEl) return;
+ const stats = countEdges(n.fp);
+ const kindLabel = n.is_self ? "SELF" : (n.distance >= 2 ? "TRANSITIVE" : "DIRECT PEER");
+ const kindCls = n.is_self ? "td-kind-host" : (n.distance >= 2 ? "td-kind-cont" : "td-kind-net");
+ const statusBadge = `This node's Ed25519 identity. The fingerprint goes into a DNS TXT record so other psyc nodes can discover this one. The public key lets them verify any feed we publish — the private key never leaves this box.
-
node fingerprint
diff --git a/src/psyc/cockpit/templates/admin_federation_network.html b/src/psyc/cockpit/templates/admin_federation_network.html
new file mode 100644
index 0000000..98487cf
--- /dev/null
+++ b/src/psyc/cockpit/templates/admin_federation_network.html
@@ -0,0 +1,51 @@
+{% extends "base.html" %}
+{% block title %}Federation network — psyc admin{% endblock %}
+{% block body_class %}wide{% endblock %}
+{% block content %}
+
+
+
Federation Network
+ {{ stats.total_peers }} direct · … transitive
+
+ Force-directed map of the federation this node sits inside. Self at the center, directly-registered peers at distance 1, peers-of-peers (fetched from each trusted peer's /federation/network) at distance 2. Edges: vouches (solid), signals (dashed, thickness ∝ 24h volume), knows (dotted grey).
+ ← federation hub · discovery · vouches · quorum · log
+
+
+
direct peers
{{ stats.total_peers }}
+
vouched / trusted
{{ stats.vouched_peers }}
+
vouches issued
{{ stats.vouches_issued }}
+
signals (24h)
{{ stats.signals_buffered_24h }}
+
distinct hashes
{{ stats.distinct_signal_hashes_24h }}
+
quorum-met
{{ stats.quorum_met_count }}
+
+
+
+
loading federation network…
+
+
+
Click any node in the graph above to inspect it.
+
Self fingerprint: {{ fingerprint }}
+
+
+
+{% endblock %}
diff --git a/src/psyc/lines/network_view.py b/src/psyc/lines/network_view.py
new file mode 100644
index 0000000..f33fc14
--- /dev/null
+++ b/src/psyc/lines/network_view.py
@@ -0,0 +1,423 @@
+"""Network view — federation graph: self + peers + edges (vouches + signals).
+
+The federation/identity layer (psyc.lines.federation) gives us node identity
+and a peer registry; discovery walks DNS-SD; vouching adds the web-of-trust;
+the signal buffer records what peers have told us. This module is the
+*visualization primitive* — it stitches those into a node+edge graph the
+cockpit renders as a force-directed map of the federation we sit in.
+
+Three views:
+ * build_local_view() — what we know first-hand: self at center, direct
+ peers around us, edges = vouches + 24h signals.
+ * build_transitive_view() — local_view plus peers-of-peers, fetched from
+ each trusted peer's /federation/network endpoint.
+ Cached aggressively to avoid spamming peers.
+ * build_public_view() — JSON-safe payload to publish at /federation/network.
+ Only TRUSTED peers + our outbound vouches; signed.
+"""
+
+from __future__ import annotations
+
+import base64
+import json
+import threading
+import time
+from datetime import datetime, timedelta, timezone
+from typing import Any, Dict, List, Optional, Tuple
+
+import httpx
+from pydantic import BaseModel, Field
+
+from psyc import db, log
+from psyc.lines import federation
+
+
+_log = log.get(__name__)
+
+
+SIGNAL_WINDOW_HOURS = 24
+TRANSITIVE_CACHE_TTL = 300.0 # 5 minutes
+TRANSITIVE_FETCH_TIMEOUT = 4.0
+
+
+# ---------- data model --------------------------------------------------
+
+class NetworkNode(BaseModel):
+ """One vertex on the federation map.
+
+ `distance` is the topological hop count from self: 0 for self, 1 for
+ directly-registered peers, 2 for peers-of-peers discovered via the
+ transitive fetch. `status` is the trust label the UI colors by.
+ """
+ fingerprint: str
+ domain: Optional[str] = None
+ label: str
+ status: str # "self" | "trusted" | "vouched" | "unknown" | "blocked"
+ is_self: bool = False
+ distance: int = 1
+
+
+class NetworkEdge(BaseModel):
+ """One edge on the federation map.
+
+ `kind` drives stroke style in the UI: vouch = solid, signal = dashed
+ flow with thickness ∝ weight, knows = dotted grey transitive hint.
+ """
+ source_fingerprint: str
+ target_fingerprint: str
+ kind: str # "vouch" | "signal" | "knows"
+ weight: float = 1.0
+ label: str = ""
+ bidirectional: bool = False
+
+
+class NetworkView(BaseModel):
+ """Renderable snapshot — nodes + edges + headline stats."""
+ nodes: List[NetworkNode] = Field(default_factory=list)
+ edges: List[NetworkEdge] = Field(default_factory=list)
+ generated_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))
+ stats: Dict[str, Any] = Field(default_factory=dict)
+
+
+# ---------- internal helpers --------------------------------------------
+
+def _short_fp(fp: str) -> str:
+ if len(fp) >= 16:
+ return f"{fp[:8]}…{fp[-8:]}"
+ return fp
+
+
+def _peer_status_label(peer: federation.Peer) -> str:
+ """Map peers.status + vouch-eligibility into the UI status taxonomy."""
+ if peer.status == "trusted":
+ return "trusted"
+ if peer.status == "blocked":
+ return "blocked"
+ # unknown — but if a quorum of trusted peers have vouched, surface as "vouched"
+ if federation.peer_is_listening_eligible(peer.fingerprint):
+ return "vouched"
+ return "unknown"
+
+
+def _signal_counts_24h() -> Tuple[Dict[str, int], int, int]:
+ """Per-peer signal count over the last `SIGNAL_WINDOW_HOURS`, plus totals.
+
+ Returns (counts_by_fingerprint, total_buffered, distinct_hashes).
+ """
+ cutoff = (datetime.now(timezone.utc) - timedelta(hours=SIGNAL_WINDOW_HOURS)).isoformat()
+ counts: Dict[str, int] = {}
+ distinct_hashes: set = set()
+ total = 0
+ # recent_signals gives newest first; one call is enough for a 24h window
+ # under any realistic volume — the buffer is small.
+ for row in db.recent_signals(limit=10_000):
+ received = str(row.get("received_at") or "")
+ if received < cutoff:
+ break
+ fp = row.get("peer_fingerprint") or ""
+ if not fp:
+ continue
+ counts[fp] = counts.get(fp, 0) + 1
+ distinct_hashes.add(row.get("signal_hash") or "")
+ total += 1
+ return counts, total, len(distinct_hashes)
+
+
+# ---------- local view --------------------------------------------------
+
+def build_local_view() -> NetworkView:
+ """Everything we know first-hand — self + direct peers + their edges to us.
+
+ Self sits at distance 0. Each row from `db.list_peers()` becomes a
+ distance-1 node, with `status` derived from `peers.status` (falling
+ back to "vouched" if a vouch quorum names them) or "blocked".
+
+ Edges:
+ * self -> peer vouch — for every peer we've issued a vouch for
+ * peer -> self vouch — for every received vouch naming our fp
+ * self -> peer signal — one per peer that has sent us ≥1 signal in 24h,
+ weight = count, label = "N signals"
+
+ Stats line up with what the admin page header shows.
+ """
+ our_fp = federation.node_fingerprint()
+ peers = federation.list_peers()
+ nodes: List[NetworkNode] = [
+ NetworkNode(
+ fingerprint=our_fp,
+ label="self",
+ status="self",
+ is_self=True,
+ distance=0,
+ )
+ ]
+ peer_by_fp: Dict[str, federation.Peer] = {}
+ for p in peers:
+ if p.fingerprint == our_fp:
+ # Defensive — should never happen, but don't double-render self.
+ continue
+ peer_by_fp[p.fingerprint] = p
+ nodes.append(NetworkNode(
+ fingerprint=p.fingerprint,
+ domain=p.domain,
+ label=p.domain or _short_fp(p.fingerprint),
+ status=_peer_status_label(p),
+ is_self=False,
+ distance=1,
+ ))
+
+ edges: List[NetworkEdge] = []
+
+ # Vouches WE have issued — self -> target.
+ out_vouch_targets: set = set()
+ for v in federation.our_vouches():
+ # Only render an edge if the target is in our peer registry
+ # OR is the federation broadcast target — but for the local map we
+ # only show edges to nodes we render, so skip strangers.
+ if v.target_fingerprint not in peer_by_fp:
+ continue
+ out_vouch_targets.add(v.target_fingerprint)
+ edges.append(NetworkEdge(
+ source_fingerprint=our_fp,
+ target_fingerprint=v.target_fingerprint,
+ kind="vouch",
+ weight=1.0,
+ label="vouched",
+ ))
+
+ # Vouches received — non-self vouches where target == our fp.
+ for v in federation.vouches_for(our_fp):
+ if v.voucher_fingerprint == our_fp:
+ continue
+ if v.voucher_fingerprint not in peer_by_fp:
+ # Voucher not in our peer table → no node to attach to. Skip;
+ # the local map is honest about what's renderable.
+ continue
+ # Mark bidirectional on the existing outbound vouch when applicable so
+ # the UI can collapse the two lines.
+ existing = next(
+ (e for e in edges
+ if e.kind == "vouch"
+ and e.source_fingerprint == our_fp
+ and e.target_fingerprint == v.voucher_fingerprint),
+ None,
+ )
+ if existing is not None:
+ existing.bidirectional = True
+ existing.label = "vouched ↔"
+ continue
+ edges.append(NetworkEdge(
+ source_fingerprint=v.voucher_fingerprint,
+ target_fingerprint=our_fp,
+ kind="vouch",
+ weight=1.0,
+ label="vouches us",
+ ))
+
+ # Signal edges — one per peer that has reported anything in 24h.
+ sig_counts, total_signals, distinct_hashes = _signal_counts_24h()
+ for fp, count in sig_counts.items():
+ if fp not in peer_by_fp:
+ # Signals only land from listening-eligible peers, but if the
+ # operator just demoted a peer the buffer can still contain
+ # rows referencing a now-deleted peer. Skip cleanly.
+ continue
+ edges.append(NetworkEdge(
+ source_fingerprint=fp,
+ target_fingerprint=our_fp,
+ kind="signal",
+ weight=float(count),
+ label=f"{count} signals/24h",
+ ))
+
+ # Quorum-met count over the signals we've seen in the window.
+ quorum_met = 0
+ seen_hashes: set = set()
+ for row in db.recent_signals(limit=10_000):
+ h = row.get("signal_hash") or ""
+ if not h or h in seen_hashes:
+ continue
+ seen_hashes.add(h)
+ if federation.is_quorum_met(h):
+ quorum_met += 1
+
+ vouched_peers = sum(
+ 1 for n in nodes
+ if not n.is_self and n.status in ("trusted", "vouched")
+ )
+ stats: Dict[str, Any] = {
+ "total_peers": len(peer_by_fp),
+ "vouched_peers": vouched_peers,
+ "signals_buffered_24h": total_signals,
+ "distinct_signal_hashes_24h": distinct_hashes,
+ "quorum_met_count": quorum_met,
+ "vouches_issued": len(out_vouch_targets),
+ }
+
+ return NetworkView(nodes=nodes, edges=edges, stats=stats)
+
+
+# ---------- transitive view ---------------------------------------------
+
+_TRANSITIVE_CACHE: Dict[str, Any] = {"ts": 0.0, "view": None}
+_TRANSITIVE_LOCK = threading.Lock()
+
+
+def _fetch_peer_network(domain: str, timeout: float = TRANSITIVE_FETCH_TIMEOUT) -> Optional[Dict[str, Any]]:
+ """GET /federation/network on a peer. Returns dict on success, None otherwise.
+
+ Failure modes are logged at info — one slow/broken peer must NEVER abort
+ the transitive walk.
+ """
+ if not domain:
+ return None
+ url = f"https://{domain}/federation/network"
+ try:
+ with httpx.Client(timeout=timeout) as client:
+ r = client.get(url)
+ r.raise_for_status()
+ data = r.json()
+ except Exception as exc: # noqa: BLE001 — every failure mode is "skip this peer"
+ _log.info("network_view.transitive.skip", domain=domain, reason=str(exc)[:120])
+ return None
+ if not isinstance(data, dict):
+ _log.info("network_view.transitive.bad_shape", domain=domain, kind=type(data).__name__)
+ return None
+ return data
+
+
+def build_transitive_view(force_refresh: bool = False) -> NetworkView:
+ """Local view + distance-2 nodes pulled from each trusted peer.
+
+ Cached for `TRANSITIVE_CACHE_TTL` seconds — peer fan-out is expensive.
+ Cache is opt-out via `force_refresh` (the admin "re-fetch" button).
+ """
+ now = time.time()
+ if not force_refresh:
+ with _TRANSITIVE_LOCK:
+ cached = _TRANSITIVE_CACHE.get("view")
+ cached_ts = float(_TRANSITIVE_CACHE.get("ts") or 0.0)
+ if cached is not None and (now - cached_ts) < TRANSITIVE_CACHE_TTL:
+ return cached
+
+ view = build_local_view()
+ our_fp = view.nodes[0].fingerprint
+ existing_fps: set = {n.fingerprint for n in view.nodes}
+
+ # Index direct peers by domain so we can attribute distance=2 to a parent.
+ direct_by_fp: Dict[str, NetworkNode] = {
+ n.fingerprint: n for n in view.nodes if not n.is_self
+ }
+ direct_trusted = [n for n in direct_by_fp.values() if n.status == "trusted" and n.domain]
+
+ transitive_count = 0
+ for parent in direct_trusted:
+ data = _fetch_peer_network(parent.domain or "")
+ if data is None:
+ continue
+ # The peer's payload mirrors build_public_view's shape:
+ # {fingerprint, peers: [{fingerprint, domain}], vouches: [{voucher, target}]}.
+ their_peers = data.get("peers") or []
+ for pp in their_peers:
+ if not isinstance(pp, dict):
+ continue
+ fp = str(pp.get("fingerprint") or "")
+ if not fp or fp == our_fp:
+ continue
+ if fp in existing_fps:
+ # Already a direct peer (or our self) — but still draw the
+ # "knows" edge from parent to it to express the topology.
+ view.edges.append(NetworkEdge(
+ source_fingerprint=parent.fingerprint,
+ target_fingerprint=fp,
+ kind="knows",
+ weight=0.5,
+ label="knows",
+ ))
+ continue
+ domain = str(pp.get("domain") or "") or None
+ view.nodes.append(NetworkNode(
+ fingerprint=fp,
+ domain=domain,
+ label=domain or _short_fp(fp),
+ status="unknown",
+ is_self=False,
+ distance=2,
+ ))
+ existing_fps.add(fp)
+ transitive_count += 1
+ view.edges.append(NetworkEdge(
+ source_fingerprint=parent.fingerprint,
+ target_fingerprint=fp,
+ kind="knows",
+ weight=0.5,
+ label="knows",
+ ))
+
+ view.stats["transitive_nodes"] = transitive_count
+
+ with _TRANSITIVE_LOCK:
+ _TRANSITIVE_CACHE["view"] = view
+ _TRANSITIVE_CACHE["ts"] = now
+ return view
+
+
+# ---------- public payload ----------------------------------------------
+
+def build_public_view() -> Dict[str, Any]:
+ """JSON-safe public payload returned from GET /federation/network.
+
+ Only TRUSTED peers leak — never unknown or blocked. Vouches we've
+ issued ride along so peers can reconstruct the WoT shape. The whole
+ payload is signed with our Ed25519 key (canonical-JSON, then base64).
+
+ Signal hashes are deliberately omitted — those are internal state and
+ would leak who's reporting what to whom.
+ """
+ our_fp = federation.node_fingerprint()
+ trusted_peers: List[Dict[str, Any]] = []
+ for p in federation.list_peers():
+ if p.status != "trusted":
+ continue
+ trusted_peers.append({
+ "domain": p.domain,
+ "fingerprint": p.fingerprint,
+ "first_seen": p.discovered_at,
+ })
+ vouches: List[Dict[str, Any]] = []
+ for v in federation.our_vouches():
+ vouches.append({
+ "voucher_fingerprint": v.voucher_fingerprint,
+ "target_fingerprint": v.target_fingerprint,
+ "issued_at": v.issued_at.isoformat(),
+ "expires_at": v.expires_at.isoformat() if v.expires_at else None,
+ })
+
+ payload: Dict[str, Any] = {
+ "version": federation.FEED_VERSION,
+ "fingerprint": our_fp,
+ "generated_at": datetime.now(timezone.utc).isoformat(),
+ "peers": trusted_peers,
+ "vouches": vouches,
+ }
+ sig = federation.sign_payload(federation.canonical_json(payload))
+ payload["signature"] = base64.b64encode(sig).decode("ascii")
+ return payload
+
+
+# ---------- admin-only payload (data endpoint) --------------------------
+
+def build_admin_view(include_transitive: bool = True) -> Dict[str, Any]:
+ """Full view for the admin data endpoint — JSON-serializable.
+
+ Unlike `build_public_view`, this DOES include unknown + blocked peers
+ and recent signal hashes — it's only ever served behind admin auth.
+ """
+ view = build_transitive_view() if include_transitive else build_local_view()
+ return {
+ "self_fingerprint": view.nodes[0].fingerprint,
+ "nodes": [n.model_dump() for n in view.nodes],
+ "edges": [e.model_dump() for e in view.edges],
+ "stats": view.stats,
+ "generated_at": view.generated_at.isoformat(),
+ }
diff --git a/tests/test_network_view.py b/tests/test_network_view.py
new file mode 100644
index 0000000..152da47
--- /dev/null
+++ b/tests/test_network_view.py
@@ -0,0 +1,334 @@
+"""Network view — local + transitive + public payload tests."""
+
+from __future__ import annotations
+
+import base64
+from datetime import datetime, timedelta, timezone
+from typing import Any, Dict
+from unittest.mock import patch
+
+import pytest
+from sqlalchemy import create_engine
+
+from psyc import db
+from psyc.lines import federation, network_view
+from psyc.lines.network_view import (
+ NetworkEdge,
+ NetworkNode,
+ NetworkView,
+ build_local_view,
+ build_public_view,
+ build_transitive_view,
+)
+
+
+# ---------- fixtures ----------------------------------------------------
+
+@pytest.fixture
+def fresh_db(tmp_path, monkeypatch):
+ test_db = tmp_path / "test.db"
+ eng = create_engine(f"sqlite:///{test_db}", future=True)
+ db._metadata.create_all(eng, checkfirst=True)
+ monkeypatch.setattr(db, "_engine", eng)
+ monkeypatch.setattr(db, "DB_PATH", test_db)
+ yield test_db
+
+
+@pytest.fixture
+def fed_dir(tmp_path, monkeypatch):
+ d = tmp_path / "federation"
+ monkeypatch.setattr(federation, "FED_DIR", d)
+ monkeypatch.setattr(federation, "PRIVATE_KEY_PATH", d / "node.key")
+ monkeypatch.setattr(federation, "PUBLIC_KEY_PATH", d / "node.pub")
+ yield d
+
+
+@pytest.fixture(autouse=True)
+def reset_transitive_cache(monkeypatch):
+ """Prevent cache bleed between tests."""
+ monkeypatch.setattr(network_view, "_TRANSITIVE_CACHE", {"ts": 0.0, "view": None})
+ yield
+
+
+def _make_peer_pubkey() -> tuple[str, str]:
+ """Return (fingerprint, pubkey_pem) for a synthetic peer keypair."""
+ import hashlib
+ from cryptography.hazmat.primitives import serialization
+ from cryptography.hazmat.primitives.asymmetric import ed25519
+ priv = ed25519.Ed25519PrivateKey.generate()
+ pub = priv.public_key()
+ pem = pub.public_bytes(
+ encoding=serialization.Encoding.PEM,
+ format=serialization.PublicFormat.SubjectPublicKeyInfo,
+ ).decode("ascii")
+ raw = pub.public_bytes(
+ encoding=serialization.Encoding.Raw,
+ format=serialization.PublicFormat.Raw,
+ )
+ fp = hashlib.sha256(raw).digest()[:16].hex()
+ return fp, pem
+
+
+# ---------- local view --------------------------------------------------
+
+def test_local_view_empty_registry_yields_only_self(fresh_db, fed_dir):
+ view = build_local_view()
+ assert isinstance(view, NetworkView)
+ assert len(view.nodes) == 1
+ self_node = view.nodes[0]
+ assert self_node.is_self is True
+ assert self_node.distance == 0
+ assert self_node.status == "self"
+ assert self_node.fingerprint == federation.node_fingerprint()
+ assert view.edges == []
+ assert view.stats["total_peers"] == 0
+ assert view.stats["vouched_peers"] == 0
+ assert view.stats["signals_buffered_24h"] == 0
+
+
+def test_local_view_one_trusted_peer_no_edges(fresh_db, fed_dir):
+ peer_fp, peer_pem = _make_peer_pubkey()
+ federation.register_peer("peer.example", peer_fp, peer_pem, status="trusted")
+ view = build_local_view()
+ assert len(view.nodes) == 2
+ peer_node = next(n for n in view.nodes if not n.is_self)
+ assert peer_node.fingerprint == peer_fp
+ assert peer_node.status == "trusted"
+ assert peer_node.distance == 1
+ assert peer_node.domain == "peer.example"
+ assert view.edges == []
+ assert view.stats["total_peers"] == 1
+ assert view.stats["vouched_peers"] == 1
+
+
+def test_local_view_outbound_vouch_creates_edge(fresh_db, fed_dir):
+ peer_fp, peer_pem = _make_peer_pubkey()
+ federation.register_peer("peer.example", peer_fp, peer_pem, status="trusted")
+ federation.issue_vouch(peer_fp, ttl_days=30)
+ view = build_local_view()
+ vouch_edges = [e for e in view.edges if e.kind == "vouch"]
+ assert len(vouch_edges) == 1
+ e = vouch_edges[0]
+ assert e.source_fingerprint == federation.node_fingerprint()
+ assert e.target_fingerprint == peer_fp
+ assert view.stats["vouches_issued"] == 1
+
+
+def test_local_view_inbound_vouch_creates_edge(fresh_db, fed_dir):
+ """Vouches received that name us as target → peer → self edge."""
+ peer_fp, peer_pem = _make_peer_pubkey()
+ federation.register_peer("peer.example", peer_fp, peer_pem, status="trusted")
+ # Insert a vouch where peer vouches FOR us, bypassing accept_vouch (which
+ # we don't need to exercise here — the question is render shape).
+ our_fp = federation.node_fingerprint()
+ now = datetime.now(timezone.utc)
+ db.upsert_vouch(dict(
+ voucher_fingerprint=peer_fp,
+ target_fingerprint=our_fp,
+ issued_at=now.isoformat(),
+ expires_at=(now + timedelta(days=30)).isoformat(),
+ signature="x" * 88,
+ ))
+ view = build_local_view()
+ vouch_edges = [e for e in view.edges if e.kind == "vouch"]
+ assert len(vouch_edges) == 1
+ e = vouch_edges[0]
+ assert e.source_fingerprint == peer_fp
+ assert e.target_fingerprint == our_fp
+
+
+def test_local_view_bidirectional_vouches_collapse(fresh_db, fed_dir):
+ peer_fp, peer_pem = _make_peer_pubkey()
+ federation.register_peer("peer.example", peer_fp, peer_pem, status="trusted")
+ federation.issue_vouch(peer_fp, ttl_days=30)
+ # And peer vouches back at us.
+ our_fp = federation.node_fingerprint()
+ now = datetime.now(timezone.utc)
+ db.upsert_vouch(dict(
+ voucher_fingerprint=peer_fp,
+ target_fingerprint=our_fp,
+ issued_at=now.isoformat(),
+ expires_at=(now + timedelta(days=30)).isoformat(),
+ signature="x" * 88,
+ ))
+ view = build_local_view()
+ vouch_edges = [e for e in view.edges if e.kind == "vouch"]
+ assert len(vouch_edges) == 1
+ assert vouch_edges[0].bidirectional is True
+
+
+def test_local_view_signal_edge_weight_is_24h_count(fresh_db, fed_dir):
+ peer_fp, peer_pem = _make_peer_pubkey()
+ federation.register_peer("peer.example", peer_fp, peer_pem, status="trusted")
+ now_iso = datetime.now(timezone.utc).isoformat()
+ # Three signals from this peer within the window.
+ for i in range(3):
+ db.record_signal(dict(
+ peer_fingerprint=peer_fp,
+ signal_type="ioc",
+ signal_id=f"1.2.3.{i}",
+ signal_hash=f"hash-{i}",
+ received_at=now_iso,
+ raw_json="{}",
+ ))
+ # One stale signal outside the window — must be ignored.
+ stale = (datetime.now(timezone.utc) - timedelta(hours=48)).isoformat()
+ db.record_signal(dict(
+ peer_fingerprint=peer_fp,
+ signal_type="ioc",
+ signal_id="9.9.9.9",
+ signal_hash="stale",
+ received_at=stale,
+ raw_json="{}",
+ ))
+ view = build_local_view()
+ sig_edges = [e for e in view.edges if e.kind == "signal"]
+ assert len(sig_edges) == 1
+ assert sig_edges[0].weight == 3.0
+ assert sig_edges[0].source_fingerprint == peer_fp
+ assert sig_edges[0].target_fingerprint == federation.node_fingerprint()
+ assert view.stats["signals_buffered_24h"] == 3
+ assert view.stats["distinct_signal_hashes_24h"] == 3
+
+
+def test_local_view_blocked_peer_renders_with_blocked_status(fresh_db, fed_dir):
+ fp, pem = _make_peer_pubkey()
+ federation.register_peer("blocked.example", fp, pem, status="blocked")
+ view = build_local_view()
+ peer = next(n for n in view.nodes if not n.is_self)
+ assert peer.status == "blocked"
+
+
+# ---------- public view + signature round-trip --------------------------
+
+def test_public_view_excludes_unknown_and_blocked(fresh_db, fed_dir):
+ fp_t, pem_t = _make_peer_pubkey()
+ fp_u, pem_u = _make_peer_pubkey()
+ fp_b, pem_b = _make_peer_pubkey()
+ federation.register_peer("trusted.example", fp_t, pem_t, status="trusted")
+ federation.register_peer("unknown.example", fp_u, pem_u, status="unknown")
+ federation.register_peer("blocked.example", fp_b, pem_b, status="blocked")
+
+ payload = build_public_view()
+ fps = {p["fingerprint"] for p in payload["peers"]}
+ assert fp_t in fps
+ assert fp_u not in fps
+ assert fp_b not in fps
+
+
+def test_public_view_signature_round_trip(fresh_db, fed_dir):
+ fp, pem = _make_peer_pubkey()
+ federation.register_peer("trusted.example", fp, pem, status="trusted")
+ federation.issue_vouch(fp, ttl_days=30)
+ payload = build_public_view()
+
+ assert "signature" in payload
+ assert payload["fingerprint"] == federation.node_fingerprint()
+
+ sig = base64.b64decode(payload["signature"])
+ unsigned = {k: v for k, v in payload.items() if k != "signature"}
+ assert federation.verify_payload(
+ federation.canonical_json(unsigned),
+ sig,
+ federation.public_key_pem(),
+ ) is True
+
+ # Vouch we issued is in the payload.
+ targets = {v["target_fingerprint"] for v in payload["vouches"]}
+ assert fp in targets
+
+
+def test_public_view_omits_signals(fresh_db, fed_dir):
+ """Public payload must not leak who's reporting what."""
+ fp, pem = _make_peer_pubkey()
+ federation.register_peer("trusted.example", fp, pem, status="trusted")
+ db.record_signal(dict(
+ peer_fingerprint=fp,
+ signal_type="ioc",
+ signal_id="1.2.3.4",
+ signal_hash="secret-hash",
+ received_at=datetime.now(timezone.utc).isoformat(),
+ raw_json="{}",
+ ))
+ payload = build_public_view()
+ # No signal-shaped fields anywhere in the payload.
+ flat = str(payload)
+ assert "secret-hash" not in flat
+ assert "signals" not in payload
+
+
+# ---------- transitive view ---------------------------------------------
+
+def test_transitive_view_adds_distance_2_nodes(fresh_db, fed_dir):
+ direct_fp, direct_pem = _make_peer_pubkey()
+ federation.register_peer("direct.example", direct_fp, direct_pem, status="trusted")
+ # The peer reports two peers of its own.
+ far_fp_a, _ = _make_peer_pubkey()
+ far_fp_b, _ = _make_peer_pubkey()
+ fake_payload: Dict[str, Any] = {
+ "fingerprint": direct_fp,
+ "peers": [
+ {"fingerprint": far_fp_a, "domain": "far-a.example"},
+ {"fingerprint": far_fp_b, "domain": "far-b.example"},
+ ],
+ "vouches": [],
+ }
+ with patch.object(network_view, "_fetch_peer_network", return_value=fake_payload):
+ view = build_transitive_view(force_refresh=True)
+
+ distances = sorted(n.distance for n in view.nodes)
+ assert 0 in distances and 1 in distances and 2 in distances
+ transitive_fps = {n.fingerprint for n in view.nodes if n.distance == 2}
+ assert far_fp_a in transitive_fps
+ assert far_fp_b in transitive_fps
+ # "knows" edges from direct peer to each transitive.
+ knows = [e for e in view.edges if e.kind == "knows"]
+ assert len(knows) == 2
+ assert all(e.source_fingerprint == direct_fp for e in knows)
+ assert view.stats["transitive_nodes"] == 2
+
+
+def test_transitive_view_failed_fetch_does_not_abort(fresh_db, fed_dir):
+ fp_a, pem_a = _make_peer_pubkey()
+ fp_b, pem_b = _make_peer_pubkey()
+ federation.register_peer("peer-a.example", fp_a, pem_a, status="trusted")
+ federation.register_peer("peer-b.example", fp_b, pem_b, status="trusted")
+
+ far_fp, _ = _make_peer_pubkey()
+
+ def fake_fetch(domain, timeout=4.0):
+ if domain == "peer-a.example":
+ return None # simulate a fetch failure
+ return {
+ "fingerprint": fp_b,
+ "peers": [{"fingerprint": far_fp, "domain": "far.example"}],
+ "vouches": [],
+ }
+
+ with patch.object(network_view, "_fetch_peer_network", side_effect=fake_fetch):
+ view = build_transitive_view(force_refresh=True)
+ # Direct nodes both present, transitive only from B.
+ assert any(n.fingerprint == fp_a for n in view.nodes)
+ assert any(n.fingerprint == fp_b for n in view.nodes)
+ assert any(n.fingerprint == far_fp and n.distance == 2 for n in view.nodes)
+ assert view.stats["transitive_nodes"] == 1
+
+
+def test_transitive_view_skips_only_unknown_peers(fresh_db, fed_dir):
+ """Unknown peers shouldn't be queried — fetcher only called for trusted."""
+ fp_unknown, pem_u = _make_peer_pubkey()
+ fp_trusted, pem_t = _make_peer_pubkey()
+ federation.register_peer("unknown.example", fp_unknown, pem_u, status="unknown")
+ federation.register_peer("trusted.example", fp_trusted, pem_t, status="trusted")
+
+ calls = []
+
+ def fake_fetch(domain, timeout=4.0):
+ calls.append(domain)
+ return None
+
+ with patch.object(network_view, "_fetch_peer_network", side_effect=fake_fetch):
+ build_transitive_view(force_refresh=True)
+
+ assert "trusted.example" in calls
+ assert "unknown.example" not in calls