src/pyrtan/federation.py · Pyrtan source

pyrtan / sourcev0.19.0 · main

77,529 bytes · last changed 2026-10-28 in Chat links: open lobbies share their strangers without following each other; release 0.17.0

1	"""Federation: hubs that follow each other, and a directory they all keep.
2
3	A hub on its own routes sentences to the services that registered with
4	it. A federated hub also follows other hubs, the way one account follows
5	another on the fediverse, and answers for their trees too. The shape is
6	ActivityPub's (an actor document, an inbox, ``Follow`` / ``Accept`` /
7	``Undo`` / ``Announce`` activities) without being an implementation of it:
8	the activities are Pyrtan's own, the signatures are HMAC on a secret the
9	two hubs share, and nothing here federates with Mastodon.
10
11	The one rule everything rests on: following is trust, in one direction.
12
13	- A hub routes sentences only to hubs it follows. Its operator chose them.
14	- A hub that follows another may ask it to run trees; the followed hub
15	serves its followers. Nobody else can ask over federation.
16	- Being followed grants nothing: a follower receives announcements and may
17	ask, but the followed hub never routes anything to it.
18	- A hub announces what its own services hold, never what it learned from
19	the hubs it follows, and does not relay: federation is one hop. A
20	request that has already crossed a hub is refused rather than forwarded
21	again, so there are no loops to detect.
22
23	How it works. Each hub has an actor document at ``GET /actor`` (its id,
24	its inbox, its name). To follow, hub A fetches B's actor, makes a secret
25	for the pair, records B as pending, and posts a ``Follow`` carrying the
26	secret to B's inbox. B checks the policy (open, or a shared federation
27	secret the Follow must be signed with), fetches A's actor to confirm the
28	inbox is A's, records A as a follower, and posts back an ``Accept`` and an
29	``Announce`` of its capabilities, both signed with the pair secret. From
30	then on B announces whenever its capabilities change and A polls B's
31	``/capabilities`` (signed) as a fallback, so B's ``link`` is up, stale or
32	down like any peer's. ``Undo`` of the Follow ends it.
33
34	Every message between two hubs is signed: ``X-Pyrtan-Actor`` names the
35	sender, ``X-Pyrtan-Date`` carries a unix timestamp within five minutes,
36	and ``X-Pyrtan-Signature`` is HMAC-SHA256 over the method, path, date and
37	the SHA-256 of the canonical JSON body. The body sent is that canonical
38	JSON, so the receiver can verify from the body it parsed.
39
40	The directory is how hubs find each other without a server. A hub
41	started ``--listed`` puts an entry for itself (id, name, the trees it
42	holds, when) in every ``Announce`` it sends; every hub keeps the entries
43	it hears and passes them on in its own announces, in both directions of
44	every follow. Nothing central holds the directory: every federated hub
45	does, and pyrtan.com publishes only static files on GitHub Pages: the
46	default network, the hub list (hubs operators added by pull request,
47	which pyrtan.com asks every hour), and an hourly snapshot of what they
48	know. No hub is run for it, and none has to be up for it. An
49	entry is a hint (a hub two follows away cannot verify who said it), kept
50	until its hub has not been heard of for a week, and routing never uses
51	it: finding a hub in the directory and following it is still the one act
52	of trust.
53
54	Networks are names and secrets. A hub can be in any number and start
55	one (``create_network``, a network of one); another joins with its
56	invite, ``name:secret@hub`` (``join``). A Follow made in a network names
57	it by ``network_id`` (from the name and the secret, so it gives neither
58	away) and is signed with its secret; the followed hub takes it only if
59	it is in that network. The directory heard over a link is kept with the
60	link's network and passed on only over links in the same one, so a
61	private network's hubs never reach the public directory, and ``GET
62	/directory`` shows them only to the operator. The default network
63	(``default_network``) has a public secret: it names the network, it does
64	not guard it, and nothing here needs it to. Its name and secret are
65	built in, so joining it asks nobody for them; pyrtan.com publishes the
66	same two values as a mirror, not as the authority.
67
68	Joining the default network tries what the hub already knows before
69	anything else: the hubs it follows in it, then the hubs its directory
70	has heard in it, most recent first. Only when none of those accepts does
71	it read the hub list and then the snapshot, and only as far as it needs.
72	So a hub that has been in the network once never needs pyrtan.com again,
73	and any hub's public ``/directory`` or any mirror of the list will do as
74	a starting point (``$PYRTAN_HUB_LIST``). A hub started ``--lan`` also
75	hears the hubs on its own network say they are there (:mod:`pyrtan.lan`)
76	and tries those right after the ones it follows: no internet needed.
77	A follow outside any network (and every member's) works as before.
78
79	Chat. A hub with a lobby (``--chat``) asks the hubs it follows, and
80	the ones it keeps a chat link with (``ChatLink``, ``--chat-reach``: for
81	chat only, never a follow), for a stranger with ``Meet``, and the two seats talk in ``Say`` and end with
82	``Part``, with ``Typing`` while someone types, both ways between any two
83	hubs that share a secret; a ``Say`` is logged without its text and a
84	``Typing`` not at all. See :mod:`pyrtan.chat`.
85
86	What this module refuses: open relaying (no hop beyond one), asymmetric
87	secrets (there is no key infrastructure to lean on in the standard
88	library, and a pair secret is enough for hubs whose operators chose each
89	other), and any routing decision that was not an operator's follow.
90	"""
91
92	from __future__ import annotations
93
94	import hashlib
95	import hmac
96	import json
97	import os
98	import re
99	import secrets as _secrets
100	import threading
101	import time
102	import uuid
103	from collections import deque
104	from dataclasses import dataclass, field
105	from pathlib import Path
106	from typing import Any, Callable, Iterable
107
108	from . import __version__
109	from .service import http_json
110	from .store import BlobStore, StoreError, check_key, digest as blob_digest
111
112	DEFAULT_HUB_LIST = "https://pyrtan.com/rendezvous/hubs.json"
113	DEFAULT_SNAPSHOT = "https://pyrtan.com/rendezvous/directory.json"
114	# The default network, built in: public on purpose, so it names the network
115	# and guards nothing. site/rendezvous/network.json publishes the same values.
116	DEFAULT_NETWORK_NAME = "pyrtan"
117	DEFAULT_NETWORK_SECRET = "public-default-network-of-pyrtan-com"
118	NETWORK_NAME = re.compile(r"^[a-z0-9][a-z0-9-]{0,39}$")
119	INVITE = re.compile(r"^([a-z0-9][a-z0-9-]{0,39}):([A-Za-z0-9_-]{16,128})(?:@(https?://\S+))?$")
120	SIGNATURE_WINDOW = 300.0 # seconds a signed message stays valid
121	MAX_HOPS = 1 # federation is one hop; raise with care
122	STALE_AFTER = 30.0
123	DOWN_AFTER = 90.0
124	ANNOUNCE_EVERY = 60.0 # seconds between announces when nothing changed: the directory's heartbeat
125	DIRECTORY_TTL = 7 * 86400 # an entry unheard of for this long is dropped
126	DIRECTORY_MAX = 500
127	HEARD_UP = 600.0 # a directory entry seen this recently reads "up"
128	HEARD_STALE = 86400.0
129	ACTIVITY_TYPES = ("Follow", "Accept", "Reject", "Undo", "Announce", "Keep", "Kept", "Fetch", "Ask", "Answer", "Meet", "Say", "Typing", "Part", "ChatLink")
130	CHAT_TYPES = ("Meet", "Say", "Typing", "Part") # a lobby's: two strangers on two hubs
131	UNLOGGED = ("Typing",) # said every couple of seconds while someone types: not worth a log line
132	ASK_TIMEOUT = 30.0 # seconds a hub waits for a member to answer an Ask
133	MEMBER_PREFIX = "urn:pyrtan:member:" # an actor with no inbox: a browser or a phone, which polls a mailbox
134	KEY_PREFIX = "pyr:" # a member may be its device key instead (_meta/identity.md §7)
135	MAILBOX_MAX = 200 # activities a hub holds for a member that has not polled
136	REPORT_TTL = 30 * 86400 # a gap report unheard of for this long is dropped
137	REPORT_MAX = 2000
138	REPORT_ID = re.compile(r"^[a-z0-9][a-z0-9-]/[a-z0-9][a-z0-9-]@\d+$") # namespace/name@version: a catalog tree
139	HOLDER_FRESH = 86400.0 # a holder confirmed within this counts toward the replicas
140	ASK_WINDOW = 600.0 # a member asked to keep within this counts as covered until it confirms
141	SIGNED_PATHS = ("/inbox", "/run", "/intent", "/resume", "/capabilities")
142
143
144	# ---------------------------------------------------------------- the hub list
145
146
147	def _read_json(source: str) -> Any:
148	"""A URL or a file, as JSON; None when it cannot be had."""
149	try:
150	if source.startswith(("http://", "https://")):
151	status, body = http_json("GET", source, timeout=10.0)
152	return body if status == 200 else None
153	return json.loads(Path(source).read_text(encoding="utf-8"))
154	except Exception:
155	return None
156
157
158	def hub_urls(body: Any) -> list[str]:
159	"""The hubs a list names: ``{"hubs": ["https://…", …]}`` (a hub list)
160	or ``{"hubs": [{"id": "https://…", "seen": …}, …]}`` (a hub's
161	``/directory``, or a snapshot), the latter most recently heard first."""
162	if not isinstance(body, dict) or not isinstance(body.get("hubs"), list):
163	return []
164	rows = body["hubs"]
165	if all(isinstance(h, dict) for h in rows):
166	rows = sorted(rows, key=lambda h: -float(h.get("seen", 0) or 0))
167	urls = [h if isinstance(h, str) else h.get("id") if isinstance(h, dict) else None for h in rows]
168	return list(dict.fromkeys(u.rstrip("/") for u in urls if isinstance(u, str) and u.startswith(("http://", "https://"))))
169
170
171	def rendezvous_hubs() -> list[str]:
172	"""Hubs to try first, from ``$PYRTAN_RENDEZVOUS`` (one hub),
173	``$PYRTAN_HUB_LIST`` (any number of URLs or files, separated by spaces
174	or commas: hub lists, a hub's ``/directory``, a snapshot or a mirror of
175	one, in order), or the hub list at pyrtan.com. Never raises: nothing
176	readable is an empty list."""
177	single = os.environ.get("PYRTAN_RENDEZVOUS")
178	if single:
179	return [single.rstrip("/")]
180	sources = os.environ.get("PYRTAN_HUB_LIST", "").replace(",", " ").split() or [DEFAULT_HUB_LIST]
181	out: list[str] = []
182	for source in sources:
183	out.extend(u for u in hub_urls(_read_json(source)) if u not in out)
184	return out
185
186
187	def snapshot_hubs() -> list[str]:
188	"""Hubs from the directory snapshot pyrtan.com bakes hourly
189	(``$PYRTAN_SNAPSHOT`` overrides the URL or names a file), most recently
190	heard first. What a new hub can try when no listed hub answers."""
191	return hub_urls(_read_json(os.environ.get("PYRTAN_SNAPSHOT") or DEFAULT_SNAPSHOT))
192
193
194	# ---------------------------------------------------------------- networks
195
196
197	def network_id(name: str, secret: str) -> str:
198	"""What a Follow names a network by: derived from its name and secret,
199	so two networks that share a name never meet, and the id gives the
200	secret away to nobody."""
201	return hashlib.sha256(f"pyrtan-network\n{name}\n{secret}".encode("utf-8")).hexdigest()[:16]
202
203
204	@dataclass
205	class Network:
206	"""A set of hubs that sign their follows with one secret. ``public``
207	marks one whose secret is published (the default network at
208	pyrtan.com): its directory is shown to anyone, a private one's only
209	to the operator."""
210
211	name: str
212	secret: str
213	public: bool = False
214
215	@property
216	def id(self) -> str:
217	return network_id(self.name, self.secret)
218
219	def invite(self, hub: str \| None = None) -> str:
220	return f"{self.name}:{self.secret}" + (f"@{hub.rstrip('/')}" if hub else "")
221
222	def to_state(self) -> dict[str, Any]:
223	return {"name": self.name, "secret": self.secret, "public": self.public}
224
225	@classmethod
226	def from_state(cls, d: Any) -> "Network \| None":
227	if not isinstance(d, dict) or not isinstance(d.get("name"), str) or not isinstance(d.get("secret"), str):
228	return None
229	if not NETWORK_NAME.match(d["name"]) or len(d["secret"]) < 16:
230	return None
231	return cls(d["name"], d["secret"], bool(d.get("public", False)))
232
233
234	def parse_invite(text: str) -> tuple[Network, str \| None]:
235	"""``name:secret`` or ``name:secret@https://hub``: a network, and the
236	hub to follow into it. Raises ValueError on anything else."""
237	m = INVITE.match(text.strip()) if isinstance(text, str) else None
238	if not m:
239	raise ValueError("an invite is NAME:SECRET or NAME:SECRET@https://hub (a name of lowercase letters, digits and dashes; a secret of at least 16 letters, digits, dashes or underscores)")
240	return Network(m.group(1), m.group(2)), (m.group(3).rstrip("/") if m.group(3) else None)
241
242
243	def default_network() -> Network \| None:
244	"""The default network: built in, so finding it asks nobody. Its secret
245	is public on purpose: it names the network rather than guarding it.
246	``$PYRTAN_NETWORK`` (a URL or a file shaped like pyrtan.com's
247	``network.json``) names another; None when that cannot be read."""
248	source = os.environ.get("PYRTAN_NETWORK")
249	if not source:
250	return Network(DEFAULT_NETWORK_NAME, DEFAULT_NETWORK_SECRET, public=True)
251	net = Network.from_state(_read_json(source))
252	if net is not None:
253	net.public = True
254	return net
255
256
257	def bootstrap_candidates() -> list[str]:
258	"""Where to look for a first hub: the hub list, then the snapshot's hubs."""
259	listed = rendezvous_hubs()
260	return listed + [h for h in snapshot_hubs() if h not in listed]
261
262
263	def first_live_hub(hubs: list[str] \| None = None) -> str \| None:
264	"""The first listed hub whose ``/actor`` answers as a hub; when none
265	does, the first hub from the snapshot that answers; else None."""
266	tried = hubs if hubs is not None else rendezvous_hubs()
267	for batch in ([tried] if hubs is not None else [tried, lambda: [h for h in snapshot_hubs() if h not in tried]]):
268	for url in (batch() if callable(batch) else batch):
269	try:
270	status, actor = http_json("GET", f"{url}/actor", timeout=5.0)
271	except Exception:
272	continue
273	if status == 200 and isinstance(actor, dict) and actor.get("type") == "Hub":
274	return url
275	return None
276
277
278	# ---------------------------------------------------------------- signing
279
280
281	def canonical(body: Any) -> bytes:
282	return json.dumps(body, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode("utf-8")
283
284
285	def sign(secret: str, actor: str, method: str, path: str, body: Any, date: float \| None = None) -> dict[str, str]:
286	date_text = str(int(date if date is not None else time.time()))
287	digest = hashlib.sha256(canonical(body) if body is not None else b"").hexdigest()
288	message = "\n".join([method.upper(), path, date_text, digest]).encode("utf-8")
289	mac = hmac.new(secret.encode("utf-8"), message, hashlib.sha256).hexdigest()
290	return {"X-Pyrtan-Actor": actor, "X-Pyrtan-Date": date_text, "X-Pyrtan-Signature": mac}
291
292
293	def verify(secret: str, method: str, path: str, body: Any, headers: dict[str, str], now: float \| None = None) -> str \| None:
294	"""None if the signature holds; otherwise what is wrong with it."""
295	date_text = headers.get("X-Pyrtan-Date", "")
296	mac = headers.get("X-Pyrtan-Signature", "")
297	if not date_text.isdigit() or not mac:
298	return "no signature"
299	if abs((now if now is not None else time.time()) - int(date_text)) > SIGNATURE_WINDOW:
300	return "signature too old"
301	expected = sign(secret, headers.get("X-Pyrtan-Actor", ""), method, path, body, int(date_text))["X-Pyrtan-Signature"]
302	return None if hmac.compare_digest(expected, mac) else "bad signature"
303
304
305	def _lower(headers: dict[str, str] \| None) -> dict[str, str]:
306	"""Header lookup is case-insensitive; keep the X-Pyrtan-* names as written."""
307	out: dict[str, str] = {}
308	for k, v in (headers or {}).items():
309	key = "-".join(p.capitalize() if p.lower() != "pyrtan" else "Pyrtan" for p in k.split("-"))
310	out[key] = v
311	return out
312
313
314	# ---------------------------------------------------------------- remotes
315
316
317	@dataclass
318	class Remote:
319	"""A hub we follow, or one that follows us."""
320
321	url: str
322	inbox: str
323	secret: str
324	name: str = ""
325	state: str = "pending" # pending \| accepted \| rejected
326	trees: list[dict[str, Any]] = field(default_factory=list)
327	last_ok: float = 0.0
328	last_error: str \| None = None
329	since: float = field(default_factory=time.time)
330	kind: str = "hub" # hub \| member
331	trusted: bool = False # a member that joined with the operator's token, or as an identity granted run: it may run the hub's trees
332	network: str \| None = None # the id of the network the follow was made in; None for a follow outside any
333	trusted_by: str \| None = None # "token" or "key": how a trusted member came to be
334	identity: str \| None = None # the root a member followed as, if it signed the follow with a key
335	device: str \| None = None # the key that signed it
336	note: dict[str, Any] \| None = None # its device note, when the key is a device
337
338	@property
339	def link(self) -> str:
340	if self.state != "accepted":
341	return "down" if self.state == "rejected" else "pending"
342	if self.last_ok == 0.0:
343	return "down" if self.last_error else "pending"
344	age = time.time() - self.last_ok
345	if age < STALE_AFTER:
346	return "up"
347	if age < DOWN_AFTER:
348	return "stale"
349	return "down"
350
351	def to_dict(self) -> dict[str, Any]:
352	return {"url": self.url, "name": self.name, "kind": self.kind, "state": self.state, "link": self.link, "trees": [t["name"] for t in self.trees], "last_ok": self.last_ok, "last_error": self.last_error, "trusted": self.trusted, "network": self.network, "identity": self.identity}
353
354	def to_state(self) -> dict[str, Any]:
355	return {"url": self.url, "inbox": self.inbox, "secret": self.secret, "name": self.name, "state": self.state, "since": self.since, "kind": self.kind, "trusted": self.trusted, "network": self.network,
356	"trusted_by": self.trusted_by, "identity": self.identity, "device": self.device, "note": self.note}
357
358	@classmethod
359	def from_state(cls, d: dict[str, Any]) -> "Remote":
360	text = lambda k: d.get(k) if isinstance(d.get(k), str) else None # noqa: E731
361	return cls(d["url"], d["inbox"], d["secret"], d.get("name", ""), d.get("state", "pending"), since=float(d.get("since", time.time())), kind=d.get("kind", "hub"), trusted=bool(d.get("trusted", False)), network=text("network"),
362	trusted_by=text("trusted_by"), identity=text("identity"), device=text("device"), note=d.get("note") if isinstance(d.get("note"), dict) else None)
363
364
365	def _shape_report(e: dict[str, Any], reporter: str, now: float) -> dict[str, Any] \| None:
366	"""A report as carried: valid or nothing. ``reporter`` is who said it
367	first (the entry's own, else who we heard it from)."""
368	tree_id, node, on = e.get("id"), e.get("node"), e.get("on")
369	if not (isinstance(tree_id, str) and REPORT_ID.match(tree_id) and isinstance(node, str) and isinstance(on, str)):
370	return None
371	who = e.get("reporter") if isinstance(e.get("reporter"), str) and e.get("reporter", "").startswith(("http://", "https://")) else reporter
372	try:
373	count = max(1, int(e.get("count", 1)))
374	last = min(float(e.get("last", now)), now)
375	except (TypeError, ValueError):
376	return None
377	if now - last >= REPORT_TTL:
378	return None
379	observed = e.get("observed")
380	if isinstance(observed, str) and len(observed) > 80:
381	observed = "<str>"
382	elif not isinstance(observed, (str, int, float, bool)) and observed is not None:
383	observed = f"<{type(observed).__name__}>"
384	return {"reporter": who.rstrip("/"), "id": tree_id, "node": node[:80], "on": on[:120], "observed": observed, "count": count, "last": last}
385
386
387	def _logged(activity: dict[str, Any]) -> dict[str, Any]:
388	"""An activity as the outbox and inbox logs keep it: a chat line without
389	its text, a chat link without its secret."""
390	obj = activity.get("object")
391	if activity.get("type") == "Say" and isinstance(obj, dict) and "text" in obj:
392	return {activity, "object": {obj, "text": f"<{len(str(obj['text']))} characters>"}}
393	if activity.get("type") == "ChatLink" and "secret" in activity:
394	return {**activity, "secret": "<secret>"}
395	return activity
396
397
398	def heard(seen: float, now: float \| None = None) -> str:
399	"""How recently a directory entry's hub announced, as gossip carried it."""
400	age = (now if now is not None else time.time()) - seen
401	return "up" if age < HEARD_UP else "stale" if age < HEARD_STALE else "down"
402
403
404	class Federation:
405	def __init__(self, hub_id: str, *, name: str = "hub", secret: str \| None = None, state: str \| Path \| None = None, open_follows: bool = True, capabilities: Callable[[], list[dict[str, Any]]] \| None = None, listed: bool = False, store: BlobStore \| str \| Path \| None = None, replicas: int = 2, auto_keys: dict[str, Callable[[], Any]] \| None = None, report_gaps: bool = False, own_reports: Callable[[], list[dict[str, Any]]] \| None = None) -> None:
406	self.id = hub_id.rstrip("/")
407	self.report_gaps = report_gaps # opt in: our own catalog trees' gaps travel with our announces
408	self.own_reports = own_reports or (lambda: []) # the hub's gaps for catalog trees, shaped for a report
409	self.reports: dict[str, dict[str, Any]] = {} # heard gap reports: (reporter, id, node, on, observed) -> entry
410	self.name = name
411	self.secret = secret or None # a shared federation secret: Follows must be signed with it
412	self.open_follows = open_follows # without a secret, whether strangers may follow
413	self.state_path = Path(state) if state else None
414	self.capabilities = capabilities or (lambda: [])
415	self.listed = listed # put ourselves in the directory we gossip
416	self.store = store if isinstance(store, BlobStore) else BlobStore(store)
417	self.replicas = max(0, int(replicas)) # how many keepers each of our blobs should have
418	self.auto_keys = dict(auto_keys or {}) # blobs kept automatically: key -> what to keep
419	self._auto_digest: dict[str, str] = {}
420	self.keeping: dict[str, dict[str, Any]] = {} # our blobs: key -> {sha256, size, updated, holders: {actor: {at, sha256}}}
421	self.mailboxes: dict[str, deque[dict[str, Any]]] = {} # activities waiting for members
422	self.handout: Callable[[], dict[str, Any]] = lambda: {} # the trees trusted members run: name -> the tree as JSON
423	self._handed: dict[str, dict[str, dict[str, Any]]] = {} # member -> key -> {sha256, at, confirmed}
424	self._asks: dict[str, dict[str, Any]] = {} # run id -> {member, event, answer}
425	self.chat: Any = None # the hub's lobby, when it keeps one (pyrtan.chat.Lobby)
426	self.lan: Any = None # what the hub hears on the LAN, when started --lan (pyrtan.lan.Lan)
427	self.ask_timeout = ASK_TIMEOUT
428	self.following: dict[str, Remote] = {}
429	self.followers: dict[str, Remote] = {}
430	self.networks: dict[str, Network] = {} # name -> network this hub is in
431	self.directory_entries: dict[str, dict[str, Any]] = {} # (network id, hub id) -> entry; see _dkey
432	self.outbox: deque[dict[str, Any]] = deque(maxlen=50)
433	self.inbox_log: deque[dict[str, Any]] = deque(maxlen=50)
434	self._announced: tuple[str, float] = ("", 0.0) # digest and time of the last downward announce
435	self._announced_up: tuple[str, float] = ("", 0.0) # likewise upward
436	self._lock = threading.RLock()
437	if self.state_path and self.state_path.exists():
438	self._load()
439	if listed:
440	self.listed = True
441
442	# ------------------------------------------------------------ actor
443
444	@property
445	def inbox(self) -> str:
446	return f"{self.id}/inbox"
447
448	def actor(self) -> dict[str, Any]:
449	return {"type": "Hub", "id": self.id, "name": self.name, "inbox": self.inbox, "mailbox": f"{self.id}/mailbox", "capabilities": f"{self.id}/capabilities", "directory": f"{self.id}/directory", "pyrtan": __version__, "follows": "open" if (self.open_follows and not self.secret) else "secret", "listed": self.listed, "replicas": self.replicas}
450
451	# ------------------------------------------------------------ state
452
453	def _save(self) -> None:
454	if not self.state_path:
455	return
456	with self._lock:
457	data = {
458	"listed": self.listed,
459	"networks": [n.to_state() for n in self.networks.values()],
460	"following": [r.to_state() for r in self.following.values()],
461	"followers": [r.to_state() for r in self.followers.values()],
462	"directory": list(self.directory_entries.values()),
463	"keeping": self.keeping,
464	"reports": list(self.reports.values()),
465	}
466	self.state_path.parent.mkdir(parents=True, exist_ok=True)
467	self.state_path.write_text(json.dumps(data, indent=2) + "\n", encoding="utf-8")
468
469	def _load(self) -> None:
470	assert self.state_path
471	try:
472	data = json.loads(self.state_path.read_text(encoding="utf-8"))
473	except (OSError, json.JSONDecodeError):
474	return
475	self.listed = bool(data.get("listed", False))
476	for d in data.get("networks", []):
477	net = Network.from_state(d)
478	if net is not None:
479	self.networks[net.name] = net
480	for d in data.get("following", []):
481	self.following[d["url"]] = Remote.from_state(d)
482	for d in data.get("followers", []):
483	self.followers[d["url"]] = Remote.from_state(d)
484	self.merge_directory(data.get("directory", []), via="state", keep_tags=True)
485	self.merge_reports(data.get("reports", []), via="state")
486	keeping = data.get("keeping")
487	if isinstance(keeping, dict):
488	self.keeping = {k: v for k, v in keeping.items() if isinstance(v, dict) and isinstance(v.get("holders"), dict)}
489
490	def set_listed(self, listed: bool) -> None:
491	with self._lock:
492	self.listed = listed
493	if not listed:
494	self.directory_entries.pop(self.id, None)
495	self._save()
496
497	# ------------------------------------------------------------ networks
498	# A network is a name and a secret its hubs sign their follows with. A
499	# hub may be in any number, starts one with create_network, and joins
500	# one with an invite (join). Every follow made in a network says so,
501	# and the directory heard over it is kept apart and passed on only
502	# within it, so a private network's hubs never reach the public one.
503	# A follow outside any network (follow(url), a member) works as before.
504
505	def network(self, name: str) -> Network:
506	net = self.networks.get(name)
507	if net is None:
508	raise ValueError(f"this hub is in no network named {name!r}")
509	return net
510
511	def network_by_id(self, nid: Any) -> Network \| None:
512	return next((n for n in self.networks.values() if n.id == nid), None) if isinstance(nid, str) else None
513
514	def add_network(self, net: Network) -> Network:
515	"""Be in ``net``. The same network again is a no-op; another one
516	under a name already taken is refused."""
517	with self._lock:
518	current = self.networks.get(net.name)
519	if current is not None:
520	if current.secret != net.secret:
521	raise ValueError(f"this hub is already in a different network named {net.name!r}; leave it first")
522	current.public = current.public or net.public
523	return current
524	self.networks[net.name] = net
525	self._save()
526	return net
527
528	def create_network(self, name: str, public: bool = False) -> Network:
529	if not isinstance(name, str) or not NETWORK_NAME.match(name):
530	raise ValueError("a network's name is lowercase letters, digits and dashes, at most 40")
531	if name in self.networks:
532	raise ValueError(f"this hub is already in a network named {name!r}")
533	return self.add_network(Network(name, _secrets.token_urlsafe(24), public))
534
535	def known_hubs(self, network: str) -> list[str]:
536	"""The hubs this one already knows in a network (by id), to try
537	before asking anyone: the ones it follows in it, then the ones it
538	heard on the LAN say they are in it (``--lan``), then the ones its
539	directory heard in it, most recently heard first."""
540	now = time.time()
541	with self._lock:
542	followed = [r.url for r in self.following.values() if r.kind == "hub" and r.network == network and r.state == "accepted"]
543	heard = sorted((e for e in self.directory_entries.values() if e.get("network") == network and e.get("kind", "hub") == "hub" and now - e["seen"] < DIRECTORY_TTL), key=lambda e: -e["seen"])
544	nearby = self.lan.hubs(network) if self.lan is not None else []
545	return [u for u in dict.fromkeys(followed + nearby + [e["id"].rstrip("/") for e in heard]) if u != self.id]
546
547	def join(self, net: Network, hubs: Iterable[str]) -> Remote \| None:
548	"""Be in ``net`` and follow the first of ``hubs`` that accepts a
549	follow in it (an iterable, so a caller can fetch more only when the
550	first ones fail). None when none does: the hub is then a network of
551	one, which others can join by following it."""
552	net = self.add_network(net)
553	for url in hubs:
554	url = url.rstrip("/")
555	if url == self.id:
556	continue
557	current = self.following.get(url)
558	if current is not None and current.network == net.id and current.state == "accepted":
559	return current
560	before = (current.network, current.state) if current is not None else None
561	try:
562	remote = self.follow(url, network=net.name)
563	except ValueError:
564	continue
565	if remote.state == "accepted":
566	return remote
567	with self._lock:
568	# Not accepted in this network: leave things as they were.
569	if before is None:
570	self.following.pop(url, None)
571	else:
572	remote.network, remote.state = before
573	self._save()
574	return None
575
576	def leave_network(self, name: str) -> bool:
577	"""Leave a network: unfollow the hubs followed in it, drop the hubs
578	that followed us in it, and forget its directory."""
579	net = self.networks.get(name)
580	if net is None:
581	return False
582	for url in [u for u, r in self.following.items() if r.network == net.id]:
583	self.unfollow(url)
584	with self._lock:
585	for url in [u for u, r in self.followers.items() if r.network == net.id]:
586	del self.followers[url]
587	for key in [k for k, e in self.directory_entries.items() if e.get("network") == net.id]:
588	del self.directory_entries[key]
589	del self.networks[name]
590	self._save()
591	return True
592
593	def networks_view(self) -> list[dict[str, Any]]:
594	"""The operator's view: each network with its invite."""
595	with self._lock:
596	return [{"name": n.name, "id": n.id, "public": n.public, "invite": n.invite(self.id), "following": sorted(u for u, r in self.following.items() if r.network == n.id), "followers": sorted(u for u, r in self.followers.items() if r.network == n.id)} for n in sorted(self.networks.values(), key=lambda n: n.name)]
597
598	# ------------------------------------------------------------ directory
599
600	@staticmethod
601	def _dkey(network: str \| None, hub_id: str) -> str:
602	return hub_id if network is None else f"{network} {hub_id}"
603
604	def own_entry(self) -> dict[str, Any] \| None:
605	if not self.listed:
606	return None
607	return {"id": self.id, "name": self.name, "kind": "hub", "trees": sorted({t.get("name") for t in self.capabilities() if isinstance(t, dict) and t.get("name")}), "seen": time.time()}
608
609	def gossip(self, network: str \| None = None) -> list[dict[str, Any]]:
610	"""What this hub tells a hub it is linked to in ``network`` (an id;
611	None for a link outside any) about the directory: itself, if
612	listed, and every live entry it heard in that same network."""
613	now = time.time()
614	with self._lock:
615	entries = [e for e in self.directory_entries.values() if e["id"] != self.id and e.get("network") == network and now - e["seen"] < DIRECTORY_TTL]
616	own = self.own_entry()
617	return ([own] if own else []) + sorted(({k: v for k, v in e.items() if k != "network"} for e in entries), key=lambda e: -e["seen"])
618
619	def _public_networks(self) -> set[str \| None]:
620	return {None} \| {n.id for n in self.networks.values() if n.public}
621
622	def directory_backup(self) -> list[dict[str, Any]]:
623	"""The directory as a blob for keepers: the public part, tagged,
624	so a private network's hubs are never handed to a keeper."""
625	now = time.time()
626	public = self._public_networks()
627	with self._lock:
628	entries = [dict(e) for e in self.directory_entries.values() if e.get("network") in public and now - e["seen"] < DIRECTORY_TTL]
629	own = self.own_entry()
630	return ([own] if own else []) + sorted(entries, key=lambda e: -e["seen"])
631
632	def merge_directory(self, entries: Any, via: str, network: str \| None = None, keep_tags: bool = False) -> int:
633	"""Take entries heard from ``via`` over a link in ``network``: a
634	newer ``seen`` replaces an older one, our own entry is never
635	overwritten, nothing from the future is believed, and the table
636	stays bounded. ``keep_tags`` is for our own copies (the state file,
637	a blob handed back): each entry keeps the network it carries, if we
638	are still in it. Returns how many entries changed."""
639	if not isinstance(entries, list):
640	return 0
641	now = time.time()
642	changed = 0
643	with self._lock:
644	known = {n.id for n in self.networks.values()}
645	for e in entries:
646	if not isinstance(e, dict) or not isinstance(e.get("id"), str) or not e["id"].startswith(("http://", "https://", MEMBER_PREFIX, KEY_PREFIX)):
647	continue
648	hub_id = e["id"].rstrip("/")
649	if hub_id == self.id:
650	continue
651	net = network
652	if keep_tags:
653	net = e.get("network") if isinstance(e.get("network"), str) else None
654	if net is not None and net not in known:
655	continue
656	try:
657	seen = min(float(e.get("seen", 0)), now)
658	except (TypeError, ValueError):
659	continue
660	if now - seen >= DIRECTORY_TTL:
661	continue
662	key = self._dkey(net, hub_id)
663	current = self.directory_entries.get(key)
664	if current is not None and current["seen"] >= seen:
665	continue
666	self.directory_entries[key] = {
667	"id": hub_id,
668	"name": str(e.get("name") or "")[:80],
669	"kind": "member" if hub_id.startswith((MEMBER_PREFIX, KEY_PREFIX)) else "hub",
670	"trees": [str(t)[:80] for t in e.get("trees", []) if isinstance(t, (str, int))][:200],
671	"seen": seen,
672	"via": via,
673	"network": net,
674	}
675	changed += 1
676	if len(self.directory_entries) > DIRECTORY_MAX:
677	for key in sorted(self.directory_entries, key=lambda k: self.directory_entries[k]["seen"])[: len(self.directory_entries) - DIRECTORY_MAX]:
678	del self.directory_entries[key]
679	if changed:
680	self._save()
681	return changed
682
683	def directory(self, private: bool = True) -> dict[str, Any]:
684	"""Every hub this one has heard of that asked to be found, this one
685	included if listed: one row per hub, naming the networks it was
686	heard in. ``link`` is how recently each announced. Without
687	``private`` (a caller without the operator's token), only what was
688	heard outside any network or in a public one."""
689	now = time.time()
690	visible = None if private else self._public_networks()
691	names = {n.id: n.name for n in self.networks.values()}
692	rows: dict[str, dict[str, Any]] = {}
693	own = self.own_entry()
694	if own:
695	rows[self.id] = {**own, "via": "self", "networks": sorted(n.id for n in self.networks.values() if private or n.public)}
696	with self._lock:
697	entries = [dict(e) for e in self.directory_entries.values() if now - e["seen"] < DIRECTORY_TTL and (visible is None or e.get("network") in visible)]
698	for e in sorted(entries, key=lambda e: -e["seen"]):
699	row = rows.get(e["id"])
700	if row is None:
701	rows[e["id"]] = row = {**{k: v for k, v in e.items() if k != "network"}, "networks": []}
702	if e.get("network") and e["network"] not in row["networks"]:
703	row["networks"].append(e["network"])
704	hubs = [{"id": r["id"], "name": r["name"], "kind": r.get("kind", "hub"), "trees": r["trees"], "seen": r["seen"], "link": heard(r["seen"], now), "via": r.get("via", ""), "networks": sorted(r["networks"])} for r in rows.values()]
705	hubs.sort(key=lambda h: (h["id"] != self.id, {"up": 0, "stale": 1, "down": 2}[h["link"]], h["name"], h["id"]))
706	shown = {nid for h in hubs for nid in h["networks"]}
707	return {"hub": self.id, "name": self.name, "listed": self.listed, "count": len(hubs), "hubs": hubs, "networks": {nid: names[nid] for nid in sorted(shown) if nid in names}}
708
709	# ------------------------------------------------------------ gap reports
710	# A hub that opted in (--report-gaps) puts the gaps its catalog trees met
711	# into its announces: the tree's id (namespace/name@version), the node,
712	# the path decided on and the value no branch covered, with a count;
713	# never the inputs or the rest of the state. Every federated hub keeps
714	# what it hears and passes it on, like the directory, so the reports
715	# reach the listed hubs and the snapshot pyrtan.com bakes, where an author
716	# reads them beside the tree. Nothing is reported for a tree without a
717	# namespace: there is nobody to report to.
718
719	def report_gossip(self) -> list[dict[str, Any]]:
720	now = time.time()
721	own = []
722	if self.report_gaps:
723	for r in self.own_reports():
724	shaped = _shape_report(r, self.id, now)
725	if shaped:
726	own.append(shaped)
727	with self._lock:
728	heard = [r for r in self.reports.values() if r["reporter"] != self.id and now - r["last"] < REPORT_TTL]
729	return own + sorted(heard, key=lambda r: -r["last"])
730
731	def merge_reports(self, entries: Any, via: str) -> int:
732	"""Take reports heard from ``via``: one row per reporter and gap, a
733	newer ``last`` replaces an older one, nothing from the future is
734	believed, old news is dropped, the table stays bounded."""
735	if not isinstance(entries, list):
736	return 0
737	now = time.time()
738	changed = 0
739	with self._lock:
740	for e in entries:
741	shaped = _shape_report(e, via, now) if isinstance(e, dict) else None
742	if shaped is None or shaped["reporter"] == self.id:
743	continue
744	key = json.dumps([shaped["reporter"], shaped["id"], shaped["node"], shaped["on"], shaped["observed"]], sort_keys=True, ensure_ascii=False)
745	current = self.reports.get(key)
746	if current is not None and (current["last"], current["count"]) >= (shaped["last"], shaped["count"]):
747	continue
748	self.reports[key] = shaped
749	changed += 1
750	if len(self.reports) > REPORT_MAX:
751	for key in sorted(self.reports, key=lambda k: self.reports[k]["last"])[: len(self.reports) - REPORT_MAX]:
752	del self.reports[key]
753	if changed:
754	self._save()
755	return changed
756
757	def reports_view(self) -> dict[str, Any]:
758	"""Every gap report this hub carries, one row per gap with the
759	counts and reporters summed, most seen first."""
760	rows: dict[str, dict[str, Any]] = {}
761	for r in self.report_gossip():
762	key = json.dumps([r["id"], r["node"], r["on"], r["observed"]], sort_keys=True, ensure_ascii=False)
763	row = rows.get(key)
764	if row is None:
765	rows[key] = {"id": r["id"], "node": r["node"], "on": r["on"], "observed": r["observed"], "count": r["count"], "reporters": 1, "last": r["last"]}
766	else:
767	row["count"] += r["count"]
768	row["reporters"] += 1
769	row["last"] = max(row["last"], r["last"])
770	out = sorted(rows.values(), key=lambda r: (-r["count"], -r["last"], r["id"]))
771	return {"hub": self.id, "reporting": self.report_gaps, "count": len(out), "reports": out}
772
773	# ------------------------------------------------------------ outgoing
774
775	def _activity(self, type_: str, object_: Any, **extra: Any) -> dict[str, Any]:
776	return {"type": type_, "id": f"{self.id}/activities/{uuid.uuid4()}", "actor": self.id, "published": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), "object": object_, **extra}
777
778	def _announcement(self, for_remote: Remote \| None = None) -> dict[str, Any]:
779	out = {"type": "Capabilities", "name": self.name, "trees": self.capabilities(), "directory": self.gossip(for_remote.network if for_remote is not None else None)}
780	reports = self.report_gossip()
781	if reports:
782	out["reports"] = reports
783	if for_remote is not None:
784	out["kept"] = [b["key"] for b in self.store.list_kept(for_remote.url)] # what we keep for them
785	return out
786
787	def deliver(self, remote: Remote, activity: dict[str, Any], secret: str \| None = None) -> tuple[bool, str \| None, Any]:
788	"""Post an activity to a remote's inbox, signed with the pair secret
789	(or the one given); for a member, leave it in its mailbox. Returns
790	(delivered, error, the inbox's answer)."""
791	if remote.inbox == "mailbox":
792	box = self.mailboxes.setdefault(remote.url, deque(maxlen=MAILBOX_MAX))
793	box.append(activity)
794	self.outbox.append({**activity, "to": remote.url, "delivered": True, "error": None, "mailbox": True})
795	return True, None, None
796	path = "/inbox"
797	headers = sign(secret or remote.secret, self.id, "POST", path, activity)
798	try:
799	status, answer = http_json("POST", remote.inbox, activity, timeout=10.0, headers=headers)
800	except Exception as e:
801	if activity.get("type") not in UNLOGGED:
802	self.outbox.append({**_logged(activity), "to": remote.url, "delivered": False, "error": str(e)})
803	return False, str(e), None
804	ok = status in (200, 202)
805	if activity.get("type") not in UNLOGGED:
806	self.outbox.append({**_logged(activity), "to": remote.url, "delivered": ok, "error": None if ok else f"{status}: {answer}"})
807	if ok and isinstance(answer, dict):
808	if activity.get("type") == "Announce":
809	# An inbox answers an Announce with its own gossip: one round trip, both directions.
810	self.merge_directory(answer.get("directory"), via=remote.url, network=remote.network)
811	self.merge_reports(answer.get("reports"), via=remote.url)
812	if activity.get("type") == "Keep" and answer.get("kept") == activity["object"].get("key"):
813	self.confirm(remote.url, activity["object"]["key"], activity["object"]["sha256"])
814	return ok, None if ok else f"inbox answered {status}: {answer}", answer
815
816	def mailbox(self, actor: str) -> list[dict[str, Any]]:
817	"""Hand a member what has been waiting for it, and forget it."""
818	box = self.mailboxes.get(actor.rstrip("/"))
819	if not box:
820	return []
821	out = list(box)
822	box.clear()
823	remote = self.followers.get(actor.rstrip("/"))
824	if remote is not None:
825	remote.last_ok = time.time()
826	remote.last_error = None
827	return out
828
829	def follow(self, url: str, network: str \| None = None) -> Remote:
830	"""Follow the hub at ``url``: fetch its actor, make a pair secret,
831	send the Follow. The remote is pending until it Accepts. In a
832	``network`` (a name this hub is in), the Follow names it and is
833	signed with its secret, and what the two hear of the directory over
834	this link stays in it."""
835	url = url.rstrip("/")
836	if url == self.id:
837	raise ValueError("a hub cannot follow itself")
838	net = self.network(network) if network is not None else None
839	try:
840	status, actor = http_json("GET", f"{url}/actor", timeout=10.0)
841	except Exception as e:
842	raise ValueError(f"{url} does not answer /actor: {e}") from e
843	if status != 200 or not isinstance(actor, dict) or actor.get("type") != "Hub" or not isinstance(actor.get("inbox"), str):
844	raise ValueError(f"{url} does not answer /actor as a Pyrtan hub ({status})")
845	with self._lock:
846	remote = self.following.get(url)
847	if remote is None:
848	remote = Remote(url, actor["inbox"], _secrets.token_hex(32), actor.get("name", ""))
849	self.following[url] = remote
850	else:
851	remote.inbox = actor["inbox"]
852	remote.name = actor.get("name", remote.name)
853	remote.state = "pending"
854	remote.network = net.id if net else None
855	self._save()
856	extra = {"network": net.id} if net else {}
857	activity = self._activity("Follow", url, secret=remote.secret, inbox=self.inbox, name=self.name, **extra)
858	# In a network the Follow is signed with its secret; under a shared
859	# federation secret with that; in the open case with the pair
860	# secret, which proves nothing yet but keeps every message signed.
861	ok, error, _ = self.deliver(remote, activity, net.secret if net else (self.secret or remote.secret))
862	if not ok:
863	remote.last_error = error
864	elif remote.state == "accepted":
865	self.announce_up(force=True) # tell it about us, and hear the directory, now
866	return remote
867
868	def unfollow(self, url: str) -> bool:
869	url = url.rstrip("/")
870	with self._lock:
871	remote = self.following.pop(url, None)
872	self._save()
873	if remote is None:
874	return False
875	self.deliver(remote, self._activity("Undo", {"type": "Follow", "actor": self.id, "object": url}))
876	return True
877
878	def _due(self, last: tuple[str, float], announcement: dict[str, Any], force: bool) -> tuple[bool, str]:
879	with self._lock:
880	# Every network's directory: a change in any is news for the links in it.
881	heard = sorted((e.get("network") or "", e["id"], e["trees"]) for e in self.directory_entries.values())
882	digest = hashlib.sha256(canonical({"trees": [{k: t.get(k) for k in ("name", "version", "triggers", "inputs", "language", "hosts")} for t in announcement["trees"]], "directory": [(e["id"], e["trees"]) for e in announcement["directory"]], "heard": heard})).hexdigest()
883	due = force or digest != last[0] or time.time() - last[1] >= ANNOUNCE_EVERY
884	return due, digest
885
886	def announce(self, force: bool = False) -> int:
887	"""Tell every follower what this hub's own services hold and what
888	the directory says: when either changed, on a heartbeat, or
889	``force``. Returns how many deliveries succeeded."""
890	announcement = self._announcement()
891	due, digest = self._due(self._announced, announcement, force)
892	if not due or not self.followers:
893	return 0
894	self._announced = (digest, time.time())
895	delivered = 0
896	for remote in list(self.followers.values()):
897	ok, error, _ = self.deliver(remote, self._activity("Announce", self._announcement(remote)))
898	if ok:
899	delivered += 1
900	remote.last_ok = time.time()
901	remote.last_error = None
902	else:
903	remote.last_error = error
904	return delivered
905
906	def announce_up(self, force: bool = False) -> int:
907	"""Tell every hub this one follows what it holds and what the
908	directory says. The followed hub routes none of it (being followed
909	grants nothing); it keeps the directory and, if this hub is
910	listed, finds it there."""
911	announcement = self._announcement()
912	due, digest = self._due(self._announced_up, announcement, force)
913	accepted = [r for r in self.following.values() if r.state == "accepted"]
914	if not due or not accepted:
915	return 0
916	self._announced_up = (digest, time.time())
917	delivered = 0
918	for remote in accepted:
919	ok, _, _ = self.deliver(remote, self._activity("Announce", self._announcement(remote)))
920	delivered += ok
921	return delivered
922
923	def refresh(self, remote: Remote) -> None:
924	"""Poll a hub we follow, signed: the fallback when its announces
925	do not arrive, and what keeps its link honest."""
926	if remote.state != "accepted":
927	return
928	path = "/capabilities"
929	try:
930	status, caps = http_json("GET", f"{remote.url}{path}", timeout=5.0, headers=sign(remote.secret, self.id, "GET", path, None))
931	except Exception as e:
932	remote.last_error = str(e)
933	return
934	if status == 200 and isinstance(caps, dict):
935	self._take_capabilities(remote, caps)
936	else:
937	remote.last_error = f"capabilities answered {status}"
938
939	def refresh_all(self) -> None:
940	for remote in list(self.following.values()):
941	self.refresh(remote)
942	self.auto()
943	self.announce_up()
944	self.announce()
945	self.replicate()
946	self.hand_out()
947
948	# ------------------------------------------------------------ keeping
949
950	def keep(self, key: str, data: Any) -> dict[str, Any]:
951	"""Keep a blob of our own and see that enough others keep it too."""
952	meta = self.store.put_own(key, data)
953	with self._lock:
954	entry = self.keeping.setdefault(key, {"holders": {}})
955	if entry.get("sha256") != meta["sha256"]:
956	entry["asked"] = {}
957	entry.update({"sha256": meta["sha256"], "size": meta["size"], "updated": meta["updated"]})
958	self._save()
959	self.replicate(only=key)
960	return self.keeping_view(key)
961
962	def confirm(self, actor: str, key: str, sha256: str) -> None:
963	with self._lock:
964	entry = self.keeping.get(key)
965	if entry is None:
966	return
967	entry["holders"][actor.rstrip("/")] = {"at": time.time(), "sha256": sha256}
968	entry.get("asked", {}).pop(actor.rstrip("/"), None)
969	self._save()
970
971	def holders(self, key: str) -> list[str]:
972	"""Who holds the current version of a blob, confirmed recently."""
973	entry = self.keeping.get(key)
974	if not entry:
975	return []
976	now = time.time()
977	return sorted(a for a, h in entry["holders"].items() if h.get("sha256") == entry.get("sha256") and now - float(h.get("at", 0)) < HOLDER_FRESH)
978
979	def keeping_view(self, key: str \| None = None) -> Any:
980	def view(k: str, e: dict[str, Any]) -> dict[str, Any]:
981	return {"key": k, "sha256": e.get("sha256"), "size": e.get("size"), "updated": e.get("updated"), "holders": self.holders(k), "wanted": self.replicas, "known": sorted(e["holders"])}
982
983	with self._lock:
984	if key is not None:
985	e = self.keeping.get(key)
986	return view(key, e) if e else None
987	return [view(k, e) for k, e in sorted(self.keeping.items())]
988
989	def keepers(self) -> list[Remote]:
990	"""Who we may ask to keep a blob: everyone we share a secret with, members first."""
991	remotes = [r for r in self.followers.values() if r.state == "accepted"] + [r for r in self.following.values() if r.state == "accepted"]
992	remotes.sort(key=lambda r: (r.kind != "member", r.link != "up", r.url))
993	return remotes
994
995	def replicate(self, only: str \| None = None, limit: int = 20) -> int:
996	"""Send ``Keep`` for every blob with fewer fresh holders than
997	``replicas``, to keepers that do not hold it, at most ``limit`` at a
998	time. Returns how many were sent."""
999	sent = 0
1000	now = time.time()
1001	for key in ([only] if only else list(self.keeping)):
1002	entry = self.keeping.get(key)
1003	blob = self.store.get_own(key)
1004	if not entry or not blob:
1005	continue
1006	asked = entry.setdefault("asked", {})
1007	# A member confirms later, through its mailbox: count it as covered
1008	# for a while rather than asking every member at once.
1009	have = set(self.holders(key)) \| {actor for actor, t in asked.items() if now - float(t) < ASK_WINDOW}
1010	# Someone new before a holder gone stale: the stale one may be gone for good.
1011	for remote in sorted(self.keepers(), key=lambda r: r.url in entry["holders"]):
1012	if len(have) >= self.replicas or sent >= limit:
1013	break
1014	if remote.url in have:
1015	continue
1016	ok, _, _ = self.deliver(remote, self._activity("Keep", {"key": key, "sha256": blob["sha256"], "size": blob["size"], "updated": blob["updated"], "data": blob["data"]}))
1017	if ok:
1018	sent += 1
1019	have.add(remote.url)
1020	if remote.inbox == "mailbox":
1021	asked[remote.url] = now
1022	return sent
1023
1024	def auto(self) -> None:
1025	"""Keep the automatic blobs (directory, gaps, ...) when they changed."""
1026	for key, fn in self.auto_keys.items():
1027	try:
1028	data = fn()
1029	except Exception:
1030	continue
1031	sha, size = blob_digest(data)
1032	if size > 256 * 1024 or self._auto_digest.get(key) == sha:
1033	continue
1034	self._auto_digest[key] = sha
1035	try:
1036	self.keep(key, data)
1037	except StoreError:
1038	continue
1039
1040	def hand_out(self, limit: int = 20) -> int:
1041	"""Give every trusted member the trees it should run (``trees/<name>``
1042	blobs), each once per version until the member confirms. Returns
1043	how many were sent."""
1044	trees = self.handout()
1045	if not trees:
1046	return 0
1047	now = time.time()
1048	sent = 0
1049	for remote in list(self.followers.values()):
1050	if remote.kind != "member" or not remote.trusted or remote.state != "accepted":
1051	continue
1052	handed = self._handed.setdefault(remote.url, {})
1053	for name, tree in trees.items():
1054	key = f"trees/{name}"
1055	raw = tree.raw
1056	sha, size = blob_digest(raw)
1057	h = handed.get(key)
1058	if h and h["sha256"] == sha and (h["confirmed"] or now - h["at"] < ASK_WINDOW):
1059	continue
1060	if sent >= limit:
1061	return sent
1062	ok, _, _ = self.deliver(remote, self._activity("Keep", {"key": key, "sha256": sha, "size": size, "updated": now, "data": raw}))
1063	if ok:
1064	handed[key] = {"sha256": sha, "at": now, "confirmed": False}
1065	sent += 1
1066	return sent
1067
1068	def ask(self, remote: Remote, tree: str, inputs: dict[str, Any], depth: int = 0, timeout: float \| None = None) -> dict[str, Any]:
1069	"""Have a trusted member run a tree: an Ask in its mailbox, then wait
1070	for its Answer (at most ``timeout`` seconds, default ``ask_timeout``;
1071	a member polls)."""
1072	timeout = self.ask_timeout if timeout is None else timeout
1073	run = uuid.uuid4().hex
1074	pending = {"member": remote.url, "event": threading.Event(), "answer": None}
1075	self._asks[run] = pending
1076	try:
1077	self.deliver(remote, self._activity("Ask", {"run": run, "tree": tree, "inputs": inputs, "depth": depth}))
1078	if not pending["event"].wait(timeout):
1079	return {"status": "error", "reason": "member-timeout", "tree": tree, "say": f"{remote.name or remote.url} did not answer within {int(timeout)} s; a member answers when it next polls."}
1080	answer = dict(pending["answer"] or {})
1081	answer.setdefault("tree", tree)
1082	return answer
1083	finally:
1084	self._asks.pop(run, None)
1085
1086	def kept_for(self, actor: str, key: str) -> dict[str, Any] \| None:
1087	"""A blob we keep for ``actor``, for it to take back."""
1088	return self.store.get_kept(actor, key)
1089
1090	def fetch(self, key: str) -> dict[str, Any]:
1091	"""Our own copy of a blob, or the copy a holder keeps: a hub answers
1092	at once, a member is asked through its mailbox and answers later."""
1093	own = self.store.get_own(key)
1094	if own:
1095	return {"status": "ok", "source": "own", "blob": own}
1096	asked: list[str] = []
1097	entry = self.keeping.get(key, {})
1098	for actor in sorted(entry.get("holders", {})):
1099	remote = self.followers.get(actor) or self.following.get(actor)
1100	if remote is None:
1101	continue
1102	if remote.inbox == "mailbox":
1103	self.deliver(remote, self._activity("Fetch", {"key": key}))
1104	asked.append(actor)
1105	continue
1106	path = f"/kept/{key}"
1107	try:
1108	status, blob = http_json("GET", f"{remote.url}{path}", timeout=10.0, headers=sign(remote.secret, self.id, "GET", path, None))
1109	except Exception:
1110	continue
1111	if status == 200 and isinstance(blob, dict) and "data" in blob:
1112	try:
1113	self.store.put_own(key, blob["data"])
1114	except StoreError:
1115	continue
1116	return {"status": "ok", "source": actor, "blob": self.store.get_own(key)}
1117	if asked:
1118	return {"status": "asked", "say": f"{len(asked)} member(s) asked through their mailbox; try again once they have answered.", "asked": asked}
1119	return {"status": "error", "reason": "not-held", "say": f"Nobody holds {key!r}."}
1120
1121	def _adopt(self, key: str, blob: Any, who: str) -> bool:
1122	"""Take a blob a keeper hands back, when we have no copy of our own."""
1123	if not isinstance(blob, dict) or "data" not in blob or self.store.get_own(key):
1124	return False
1125	try:
1126	self.store.put_own(key, blob["data"])
1127	except StoreError:
1128	return False
1129	with self._lock:
1130	entry = self.keeping.setdefault(key, {"holders": {}})
1131	meta = self.store.get_own(key) or {}
1132	entry.update({"sha256": meta.get("sha256"), "size": meta.get("size"), "updated": meta.get("updated")})
1133	entry["holders"][who] = {"at": time.time(), "sha256": meta.get("sha256")}
1134	self._save()
1135	if key == "directory":
1136	self.merge_directory(blob.get("data"), via=who, keep_tags=True)
1137	return True
1138
1139	def _take_capabilities(self, remote: Remote, caps: dict[str, Any]) -> None:
1140	with self._lock:
1141	# Only what the hub holds itself: a tree it learned from a hub it
1142	# follows is two hops away and not ours to route to.
1143	remote.trees = [t for t in caps.get("trees", []) if isinstance(t, dict) and not t.get("via")]
1144	remote.name = caps.get("name") or remote.name
1145	remote.last_ok = time.time()
1146	remote.last_error = None
1147
1148	# ------------------------------------------------------------ incoming
1149
1150	def secret_for(self, actor: str) -> str \| None:
1151	"""The secret a request from ``actor`` is signed with. A hub asks the
1152	hubs it follows, so the pair in which the actor follows us comes
1153	first; with a mutual follow there are two pairs and this is the one
1154	the actor signs requests to us with."""
1155	actor = actor.rstrip("/")
1156	remote = self.followers.get(actor) or self.following.get(actor)
1157	return remote.secret if remote else None
1158
1159	def verify_request(self, method: str, path: str, body: Any, headers: dict[str, str] \| None) -> str \| None:
1160	"""For a signed request to a routed path: the actor it is from, or
1161	None (and the caller answers 401)."""
1162	h = _lower(headers)
1163	actor = h.get("X-Pyrtan-Actor", "").rstrip("/")
1164	secret = self.secret_for(actor)
1165	if not actor or secret is None:
1166	return None
1167	return actor if verify(secret, method, path, body, h) is None else None
1168
1169	def handle_inbox(self, activity: Any, headers: dict[str, str] \| None, trusted: bool = False, *, trusted_by: str \| None = None, signer: Any = None) -> tuple[int, dict[str, Any]]:
1170	"""``trusted`` says the request also carried the operator's token, or
1171	a key whose root this hub grants ``run`` (``trusted_by`` says which):
1172	a member that follows that way may run the hub's trees. ``signer`` is
1173	the :class:`pyrtan.identity.Signer` of a key-signed request, if any."""
1174	h = _lower(headers)
1175	if not isinstance(activity, dict) or activity.get("type") not in ACTIVITY_TYPES or not isinstance(activity.get("actor"), str):
1176	return 400, {"status": "error", "reason": "bad-activity", "say": f"Send an activity of type {', '.join(ACTIVITY_TYPES)} with an actor."}
1177	actor = activity["actor"].rstrip("/")
1178	if h.get("X-Pyrtan-Actor", "").rstrip("/") != actor:
1179	return 401, {"status": "error", "reason": "unsigned", "say": "The signature's actor must be the activity's actor."}
1180	kind = activity["type"]
1181	if activity.get("type") not in UNLOGGED:
1182	self.inbox_log.append({**_logged(activity), "received": time.time()})
1183	if kind == "Follow":
1184	return self._on_follow(actor, activity, h, trusted, trusted_by=trusted_by, signer=signer)
1185	if kind == "ChatLink":
1186	return self._on_chat_link(actor, activity, h)
1187	followed = self.following.get(actor)
1188	follower = self.followers.get(actor)
1189	if kind in ("Announce", "Keep", "Kept", "Fetch", "Answer", *CHAT_TYPES):
1190	# Announce from a hub we follow: its trees are ours to route to;
1191	# from a follower: only the directory it carries. Keeping and
1192	# chatting go both ways between any two that share a secret.
1193	remote = followed or follower
1194	else:
1195	remote = followed if kind in ("Accept", "Reject") else follower
1196	linked = False
1197	if remote is None and kind in CHAT_TYPES and self.chat is not None:
1198	# A chat link: chat may travel on it, and nothing else does.
1199	remote = self.chat.link_for(actor)
1200	linked = remote is not None
1201	if remote is None:
1202	return 403, {"status": "error", "reason": "unknown-actor", "say": f"{actor} is not a hub this one follows or is followed by."}
1203	problem = verify(remote.secret, "POST", "/inbox", activity, h)
1204	if problem and kind in ("Announce", "Keep", "Kept", "Fetch", "Answer", *CHAT_TYPES) and followed and follower and followed is not follower:
1205	# Mutual follow: the announce may be signed with the other pair.
1206	remote = follower if remote is followed else followed
1207	problem = verify(remote.secret, "POST", "/inbox", activity, h)
1208	if problem and kind in CHAT_TYPES and self.chat is not None and not linked:
1209	# A follow made after a chat link: the line may be signed with the link's secret.
1210	link = self.chat.link_for(actor)
1211	if link is not None and verify(link.secret, "POST", "/inbox", activity, h) is None:
1212	remote, linked, problem = link, True, None
1213	if problem:
1214	return 401, {"status": "error", "reason": "unsigned", "say": problem}
1215	if kind in CHAT_TYPES:
1216	if self.chat is None:
1217	return 404, {"status": "error", "reason": "no-chat", "say": f"{self.id} keeps no lobby."}
1218	if remote.kind != "hub":
1219	return 403, {"status": "error", "reason": "not-a-hub", "say": "Strangers meet hub to hub; a member chats through its hub's lobby."}
1220	remote.last_ok = time.time()
1221	return self.chat.on_activity(kind, remote, activity.get("object"), follower=remote is follower or linked)
1222	if kind == "Accept":
1223	with self._lock:
1224	remote.state = "accepted"
1225	remote.last_ok = time.time()
1226	remote.last_error = None
1227	self._save()
1228	return 200, {"status": "ok"}
1229	if kind == "Reject":
1230	with self._lock:
1231	remote.state = "rejected"
1232	remote.last_error = str(activity.get("summary") or "the hub refused the follow")
1233	self._save()
1234	return 200, {"status": "ok"}
1235	if kind == "Announce":
1236	obj = activity.get("object")
1237	if not isinstance(obj, dict) or not isinstance(obj.get("trees"), list):
1238	return 400, {"status": "error", "reason": "bad-activity", "say": "An Announce carries {\"trees\": [...]}."}
1239	if remote is followed:
1240	self._take_capabilities(remote, obj)
1241	else:
1242	with self._lock:
1243	remote.name = obj.get("name") or remote.name
1244	remote.last_ok = time.time()
1245	remote.last_error = None
1246	if remote.kind == "member" and remote.trusted:
1247	# A trusted member runs the trees we handed it, and only those;
1248	# what each tree answers to is our copy's word, not the member's.
1249	handed = self.handout()
1250	names = {t.get("name") for t in obj.get("trees", []) if isinstance(t, dict)}
1251	remote.trees = [handed[n].capability() for n in sorted(names & set(handed))]
1252	taken = self.merge_directory(obj.get("directory", []), via=actor, network=remote.network)
1253	self.merge_reports(obj.get("reports", []), via=actor)
1254	# What it keeps for us that we no longer have: ask for it back.
1255	wanted = [k for k in obj.get("kept", []) if isinstance(k, str) and not self.store.get_own(k)]
1256	for k in wanted[:20]:
1257	self.deliver(remote, self._activity("Fetch", {"key": k}))
1258	return 200, {"status": "ok", "trees": len(remote.trees) if remote is followed else 0, "merged": taken, "directory": self.gossip(remote.network), "reports": self.report_gossip(), "kept": [b["key"] for b in self.store.list_kept(actor)]}
1259	if kind == "Keep":
1260	obj = activity.get("object")
1261	if not isinstance(obj, dict) or "data" not in obj:
1262	return 400, {"status": "error", "reason": "bad-activity", "say": "A Keep carries {\"key\", \"sha256\", \"data\"}."}
1263	try:
1264	meta = self.store.put_kept(actor, obj.get("key"), obj["data"], sha256=obj.get("sha256"), updated=obj.get("updated"))
1265	except StoreError as e:
1266	return 413, {"status": "error", "reason": "not-kept", "say": str(e)}
1267	remote.last_ok = time.time()
1268	return 200, {"status": "ok", "kept": meta["key"], "sha256": meta["sha256"]}
1269	if kind == "Kept":
1270	obj = activity.get("object")
1271	if not isinstance(obj, dict) or not isinstance(obj.get("key"), str):
1272	return 400, {"status": "error", "reason": "bad-activity", "say": "A Kept names a key."}
1273	if obj["key"].startswith("trees/"):
1274	with self._lock:
1275	handed = self._handed.setdefault(actor, {}).get(obj["key"])
1276	if handed and handed.get("sha256") == obj.get("sha256"):
1277	handed["confirmed"] = True
1278	elif "data" in obj:
1279	self._adopt(obj["key"], obj, actor)
1280	elif isinstance(obj.get("sha256"), str):
1281	self.confirm(actor, obj["key"], obj["sha256"])
1282	remote.last_ok = time.time()
1283	return 200, {"status": "ok"}
1284	if kind == "Answer":
1285	obj = activity.get("object")
1286	run = obj.get("run") if isinstance(obj, dict) else None
1287	pending = self._asks.get(run) if isinstance(run, str) else None
1288	if pending is None or pending["member"] != actor:
1289	return 404, {"status": "error", "reason": "no-such-ask", "say": "No run waits for this answer from this member."}
1290	pending["answer"] = obj.get("outcome") if isinstance(obj.get("outcome"), dict) else {"status": "error", "reason": "bad-answer", "say": "The member's answer was not an outcome."}
1291	pending["event"].set()
1292	remote.last_ok = time.time()
1293	return 200, {"status": "ok"}
1294	if kind == "Fetch":
1295	obj = activity.get("object")
1296	key = obj.get("key") if isinstance(obj, dict) else None
1297	blob = self.store.get_kept(actor, key) if isinstance(key, str) else None
1298	if blob is None:
1299	return 404, {"status": "error", "reason": "not-kept", "say": f"Nothing kept here for {actor} under {key!r}."}
1300	return 200, {"status": "ok", "blob": blob}
1301	if kind == "Undo":
1302	self.drop_follower(actor)
1303	return 200, {"status": "ok"}
1304	return 400, {"status": "error", "reason": "bad-activity"}
1305
1306	def drop_follower(self, actor: str) -> bool:
1307	"""Forget a follower, its mailbox, what we handed it and what we keep for it."""
1308	with self._lock:
1309	gone = self.followers.pop(actor, None) is not None
1310	self.mailboxes.pop(actor, None)
1311	self._handed.pop(actor, None)
1312	self.store.drop_kept(actor)
1313	self._save()
1314	return gone
1315
1316	def _on_chat_link(self, actor: str, activity: dict[str, Any], h: dict[str, str]) -> tuple[int, dict[str, Any]]:
1317	"""A hub asks for a chat link: checked as an open ``Follow`` is (signed
1318	with the pair secret it carries; its inbox is the one its ``/actor``
1319	names), then the lobby decides. A closed federation links with nobody."""
1320	pair, inbox = activity.get("secret"), activity.get("inbox")
1321	if not isinstance(pair, str) or len(pair) < 16 or not isinstance(inbox, str) or not actor.startswith(("http://", "https://")):
1322	return 400, {"status": "error", "reason": "bad-activity", "say": "A ChatLink comes from a hub and carries a pair secret and its inbox."}
1323	if self.chat is None:
1324	return 404, {"status": "error", "reason": "no-chat", "say": f"{self.id} keeps no lobby."}
1325	if self.secret or not self.open_follows:
1326	return 403, {"status": "error", "reason": "closed", "say": "This hub links only with hubs its operator chose."}
1327	problem = verify(pair, "POST", "/inbox", activity, h)
1328	if problem:
1329	return 401, {"status": "error", "reason": "unsigned", "say": problem}
1330	try:
1331	status, doc = http_json("GET", f"{actor}/actor", timeout=10.0)
1332	except Exception as e:
1333	return 400, {"status": "error", "reason": "no-actor", "say": f"{actor}/actor did not answer: {e}"}
1334	if status != 200 or not isinstance(doc, dict) or doc.get("id", "").rstrip("/") != actor or doc.get("inbox") != inbox:
1335	return 400, {"status": "error", "reason": "no-actor", "say": f"{actor}/actor does not describe this hub."}
1336	return self.chat.accept_link(Remote(actor, inbox, pair, str(activity.get("name") or doc.get("name") or "")[:80], state="accepted", last_ok=time.time(), kind="hub"))
1337
1338	def _on_follow(self, actor: str, activity: dict[str, Any], h: dict[str, str], trusted: bool = False, *, trusted_by: str \| None = None, signer: Any = None) -> tuple[int, dict[str, Any]]:
1339	pair = activity.get("secret")
1340	inbox = activity.get("inbox")
1341	if not isinstance(pair, str) or len(pair) < 16 or not isinstance(inbox, str):
1342	return 400, {"status": "error", "reason": "bad-activity", "say": "A Follow carries a pair secret and the follower's inbox."}
1343	nid = activity.get("network")
1344	net = self.network_by_id(nid)
1345	if nid is not None:
1346	# A follow in a network: signed with its secret, and only into a network this hub is in.
1347	if net is None:
1348	return 403, {"status": "error", "reason": "not-in-network", "say": "This hub is not in the network this follow names."}
1349	problem = verify(net.secret, "POST", "/inbox", activity, h)
1350	if problem:
1351	return 401, {"status": "error", "reason": "unsigned", "say": f"A follow in {net.name} is signed with its secret: {problem}."}
1352	elif self.secret:
1353	problem = verify(self.secret, "POST", "/inbox", activity, h)
1354	if problem:
1355	return 401, {"status": "error", "reason": "unsigned", "say": f"This hub takes follows signed with its federation secret: {problem}."}
1356	elif not self.open_follows:
1357	return 403, {"status": "error", "reason": "closed", "say": "This hub does not take follows."}
1358	else:
1359	problem = verify(pair, "POST", "/inbox", activity, h)
1360	if problem:
1361	return 401, {"status": "error", "reason": "unsigned", "say": problem}
1362	if inbox == "mailbox":
1363	# A member: a browser or a phone, with no address of its own. It
1364	# polls its mailbox here, signed, instead of being posted to.
1365	if actor.startswith(KEY_PREFIX):
1366	if signer is None or signer.key != actor:
1367	return 401, {"status": "error", "reason": "unsigned", "say": f"A member whose id is a key signs its follow with that key (X-Pyrtan-Key: {actor})."}
1368	elif not actor.startswith(MEMBER_PREFIX) or len(actor) < len(MEMBER_PREFIX) + 16:
1369	return 400, {"status": "error", "reason": "bad-activity", "say": f"A member's id is {MEMBER_PREFIX}<at least 16 characters>, or its key, pyr:…."}
1370	name = str(activity.get("name") or "")[:80]
1371	kind = "member"
1372	else:
1373	# The inbox must be the actor's own: ask the actor.
1374	try:
1375	status, doc = http_json("GET", f"{actor}/actor", timeout=10.0)
1376	except Exception as e:
1377	return 400, {"status": "error", "reason": "no-actor", "say": f"{actor}/actor did not answer: {e}"}
1378	if status != 200 or not isinstance(doc, dict) or doc.get("id", "").rstrip("/") != actor or doc.get("inbox") != inbox:
1379	return 400, {"status": "error", "reason": "no-actor", "say": f"{actor}/actor does not describe this follower."}
1380	name = str(activity.get("name") or doc.get("name") or "")
1381	kind = "hub"
1382	with self._lock:
1383	remote = Remote(actor, inbox, pair, name, state="accepted", last_ok=time.time(), kind=kind, trusted=bool(trusted and kind == "member"), network=net.id if net else None)
1384	if remote.trusted:
1385	remote.trusted_by = trusted_by or "token"
1386	if signer is not None and kind == "member":
1387	remote.identity, remote.device, remote.note = signer.root, signer.key, signer.note
1388	self.followers[actor] = remote
1389	self._handed.pop(actor, None)
1390	self._save()
1391	self.deliver(remote, self._activity("Accept", {"type": "Follow", "actor": actor, "object": self.id}))
1392	self.deliver(remote, self._activity("Announce", self._announcement(remote)))
1393	if remote.trusted:
1394	self.hand_out()
1395	return 202, {"status": "ok", "say": f"{self.id} accepts {actor}.", "mailbox": f"{self.id}/mailbox" if kind == "member" else None, "trusted": remote.trusted, "identity": remote.identity}
1396
1397	# ------------------------------------------------------------ views
1398
1399	def to_dict(self) -> dict[str, Any]:
1400	with self._lock:
1401	return {"id": self.id, "listed": self.listed, "following": [r.to_dict() for r in self.following.values()], "followers": [r.to_dict() for r in self.followers.values()], "directory": len({e["id"] for e in self.directory_entries.values()}) + (1 if self.listed else 0), "keeping": len(self.keeping), "store": self.store.usage(), "replicas": self.replicas, "networks": [{"name": n.name, "id": n.id, "public": n.public} for n in self.networks.values()]}