Members: a browser or a phone joins a hub through a mailbox and keeps its blobs; release 0.5.0
Claude · 2026-10-25 · fb2ff04cf84acb614edb6e31ef607ebb876d7a6a · parents 089ae82
A device that cannot be reached joins a hub the other way round. A member has an id urn:pyrtan:member:<hex> and a pair secret of its own, follows with inbox "mailbox", and from then on everything the hub would post to it waits in a bounded mailbox it polls, signed; it posts its own activities to the hub's inbox like any hub. Same signatures, same one-way trust: a member carries and gossips the directory, is listed in it as a member if it wants to be, and is never routed to.
Keeping: a blob is a JSON value under a key, at most 256 KB; a hub keeps its own (--store DIR) and keeps for others, 200 keys and 64 MB per actor. Keep hands a blob to a keeper, Kept confirms it (a hub in the same round trip, a member on its next poll), Fetch asks for it back, and every Announce names the keys the sender keeps for the receiver, so a hub that came back without its store gets its blobs from its members' next announces without anyone doing anything. --replicas N is how many fresh holders each blob should have; the poll sends Keep to keepers that do not hold it, members first, someone new before a stale holder. The directory, the gaps and the unanswered sentences are kept automatically; the state file, with the pair secrets, never is. pyrtan keep, kept and fetch drive it; /keep, /fetch, /mailbox and /kept/<key> are the paths.
The console is the member: a Member tab makes the identity, follows, polls every twenty seconds, keeps blobs in IndexedDB and hands them back, with SHA-256 and HMAC written out in the page because crypto.subtle is not available on a plain http:// LAN address. A web app manifest and a service worker make it installable and offline-capable, served by every hub and service and written beside the exported console. Design in _meta/members.md; members do not run trees yet.
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01YChr78vDHtQJ15HNbsaRLZ
Changes
README.md | 28 ++++- _meta/design.md | 13 ++- _meta/federation.md | 10 +- _meta/members.md | 136 ++++++++++++++++++++++ deploy/openmediavault/docker-compose.yml | 2 + pyproject.toml | 2 +- site/README.md | 2 +- src/pyrtan/__init__.py | 2 +- src/pyrtan/catalog.py | 2 +- src/pyrtan/cli.py | 92 ++++++++++++++- src/pyrtan/federation.py | 316 ++++++++++++++++++++++++++++++++++++++++++++------- src/pyrtan/hub.py | 54 +++++++-- src/pyrtan/page.py | 207 ++++++++++++++++++++++++++++++++- src/pyrtan/service.py | 22 +++- src/pyrtan/store.py | 187 ++++++++++++++++++++++++++++++ src/pyrtan/transport.py | 2 +- tests/test_federation.py | 186 ++++++++++++++++++++++++++++++ uv.lock | 2 +- 18 files changed, 1189 insertions(+), 76 deletions(-)
diff --git a/README.md b/README.md index d9ec28a..94647bd 100644 --- a/README.md +++ b/README.md@@ -125,6 +125,7 @@ whole format is in [`_meta/tree-format.md`](_meta/tree-format.md).| `pyrtan match "…" --trees DIR` | How a sentence reads, without running anything: the rewrites, every trigger's verdict, the winner. The tool for writing triggers. | | `pyrtan serve DIR --port 8701` | Hosts the trees as a service: `/health`, `/capabilities`, `/intent`, `/run`, `/resume`, `/gaps`, `/unanswered`. Reloads trees when their files change. `--gaps FILE` keeps gaps across restarts; `--egress public` refuses private addresses. | | `pyrtan hub --port 8700` | Routes between services. Services register with `--hub URL`; peers carry a `link` of `up`, `stale` or `down`. `--trees DIR` holds trees in the hub itself; `--federate` lets other hubs follow it and it follow them; `--listed` puts it in the directory. |+| `pyrtan keep KEY [FILE] --at HUB` | Gives your hub a blob to keep; its keepers get copies. `kept` shows who holds what; `fetch KEY` gets a blob back from your hub or whoever holds it. || `pyrtan follow [URL] --at HUB` | Makes your hub follow another (default: the first live seed from pyrtan.com): it then routes sentences to that hub's trees too. `--listed` also puts your hub in the directory. `unfollow` undoes it; `peers` lists services, hubs followed and followers; `directory` lists every hub that asked to be found, as the swarm knows it. | | `pyrtan ask "…" --at URL` | Answers a sentence through a running service or hub. | | `pyrtan gaps --at URL` | Every decision that met a value with no branch: the branches to write next. |@@ -195,7 +196,25 @@ hourly snapshot of what they know; `pyrtan follow` with no URL takesthe first seed that answers, then the first hub from the snapshot. An entry is a hint, not a route: to use a hub you found, follow it. A hub can also hold trees itself, `pyrtan hub --trees DIR`, so a whole federated swarm can be one-process on one box. The answer to a federated+process on one box.++## Members, and what they keep++A browser or a phone cannot be reached, so it joins a hub the other way+round: the console's **Member** tab makes an identity for the device,+follows the hub with it, and polls a mailbox the hub keeps for it. A+member carries the directory like any hub, is listed in it if it wants+to be, and **keeps blobs**: the hub hands its own blobs (its directory,+its gaps, its unanswered sentences, anything given to `pyrtan keep KEY+FILE`) to the members and hubs that follow it, until `--replicas` of+them have confirmed, and asks for them back when it needs them. When the+hub's box dies and a new one comes up with the same name, its members'+next announce says what they kept and the hub takes it back; `pyrtan+kept` shows who holds what, `pyrtan fetch KEY` gets a blob back from+whoever has it. Install the console to the home screen and it is the+phone app, offline-capable. Nothing secret is ever a blob. The design is+in [`_meta/members.md`](_meta/members.md). Members do not run trees yet;+that is the JavaScript runtime of v0.6. The answer to a federatedsentence carries `via: [the hub it crossed]`, and the console's Peers tab shows who follows whom. The design is in [`_meta/federation.md`](_meta/federation.md).@@ -340,9 +359,10 @@ site/ pyrtan.com: the landing page, the seed list, the catalog## Status-0.4.0. The core, the service, the hub, the command, the console, the-catalog client, the language rules, federation between hubs and the-gossiped directory are built and tested; HTTP is the one form of I/O the core speaks; a swarm can+0.5.0. The core, the service, the hub, the command, the console, the+catalog client, the language rules, federation between hubs, the+gossiped directory, members (a browser or a phone as part of the swarm)+and the blobs they keep are built and tested; HTTP is the one form of I/O the core speaks; a swarm canrequire a token, keep its gaps, refuse private addresses, and follow other swarms. Not yet built: further adapters (a browser for pages that need a script to render; files, shells, serial lines, message queues), diff --git a/_meta/design.md b/_meta/design.md index 1e504f7..17aca24 100644 --- a/_meta/design.md +++ b/_meta/design.md@@ -286,17 +286,24 @@ Five trees, all against real public APIs, all tested offline:other by is kept and gossiped by every federated hub, with a static seed list at pyrtan.com in place of a rendezvous hub; a hub can hold trees itself (`--trees`), so a federated swarm can be one process.-5. **v0.5: more forms of I/O.** First the browser: `observe` with+5. **v0.5, built: members and what they keep.** A browser or a phone+ joins a hub through a mailbox, carries the directory, and keeps the+ hub's blobs so a dead box loses nothing; the console installs as the+ phone app. Design in [`members.md`](members.md).+6. **v0.6: trees in the browser.** `pyrtan.js`, a second runtime for the+ same tree files, with the Python fixtures as shared conformance+ vectors, so a member answers sentences through its hub's mailbox.+7. **v0.7: more forms of I/O.** First the browser: `observe` with`browser = { url, wait_for, actions }` on Playwright as an optional extra, for pages that need a script to render, with the same host rule and the same extraction. Then file, shell, serial and queue adapters, each a node source with its own allow-list (paths, commands, devices, topics), so a tree reads a log, runs a named command or polls a device by the same rules it reads a page, and the trace reads the same.-6. **v0.6: the catalog grows up.** Namespaces, publishing through pull+8. **v0.8: the catalog grows up.** Namespaces, publishing through pullrequests with the checker as the gate, and opt-in gap reporting from hubs back to a tree's author, carried as another federation topic.-7. **Beside the swarm, never inside it:** an authoring aid that reads gaps+9. **Beside the swarm, never inside it:** an authoring aid that reads gapsand drafts the branch for a person to accept. ## 7. Open questions diff --git a/_meta/federation.md b/_meta/federation.md index 3d6fbd4..c5be390 100644 --- a/_meta/federation.md +++ b/_meta/federation.md@@ -245,7 +245,15 @@ because nobody can reach its inbox. To be routable it needs an address:a port forward behind TLS, or a tunnel (Cloudflare Tunnel, Tailscale Funnel) that costs nothing and opens no port. The deploy guide says how.-## 8. What this design does not claim+## 8. Members++A browser or a phone cannot be followed, because nobody can reach it. It+can follow. v0.5 adds *members*: actors with no inbox, which poll a+mailbox the hub keeps for them, and the blobs hubs and members keep for+each other. Both are in [`members.md`](members.md); the rules above+hold unchanged for them.++## 9. What this design does not claimIt does not claim a follower cannot see what the followed hub holds; that is the point of following. It does not claim a hub behind a federation diff --git a/_meta/members.md b/_meta/members.md new file mode 100644 index 0000000..e8dbae7 --- /dev/null +++ b/_meta/members.md@@ -0,0 +1,136 @@+# Design — Members: a browser or a phone as part of the swarm, and what they keep++| | |+|---|---|+| **Status** | **Built, v0.5.** Members, mailboxes, `Keep`/`Kept`/`Fetch`, the blob store, the console as an installable member; tests with Python members and a browser driven end to end. Members do not run trees; that is v0.6. |+| **Date** | 2026-10-25 |+| **Decision** | A device that cannot be reached joins a hub by following it and polling a mailbox the hub keeps for it. It carries the directory like a hub, and it keeps the hub's blobs, so that a swarm's state lives on the devices of the people who run it and a dead box loses nothing. Nothing secret is ever a blob. |+| **Where** | `src/pyrtan/federation.py` (members, mailboxes, keeping), `src/pyrtan/store.py` (blobs), `src/pyrtan/page.py` (the member in the console, the manifest, the service worker). |++## 0. The brief++*"Browsers and phone apps as part of the cluster, and distributed storage."*+Read as: the things people already carry should be nodes, and the swarm+should not depend on any one box for its memory.++## 1. What a browser can and cannot be++A tab or a phone app can make outbound HTTP, store a few hundred+megabytes locally, run JavaScript and hold a secret. It **cannot+listen**: nobody can reach it. In a browser it can only fetch hosts that+send CORS headers; a phone app wrapping the same page can fetch anything.++So a member is a federation actor without an inbox, and everything in+this design follows from that one fact.++## 2. Members++A member has an id `urn:pyrtan:member:<hex>` and a pair secret it makes+itself. It follows a hub with the ordinary `Follow`, `inbox: "mailbox"`.+The hub accepts under the same policy as any follow (open, or signed+with the federation secret), records the member with `kind: member`,+and from then on everything it would post to the member's inbox goes+into a bounded mailbox instead (`MAILBOX_MAX`, two hundred activities;+older ones fall off). The member polls `GET /mailbox`, signed, which+hands over what is waiting and forgets it; it posts its own activities+to the hub's `/inbox` like any hub. Same signatures, same one-way trust:+a member may ask the hub it follows, carries and gossips the directory,+is listed in it if it wants to be (`kind: member`), and is never routed+to.++The console is the member. Its Member tab makes the identity (kept in+local storage, one per hub origin), follows, and polls every twenty+seconds while the page is open, announcing upward every minute. The+page carries its own SHA-256 and HMAC, because `crypto.subtle` is not+available on a plain `http://` address on a LAN, which is where most+hubs are. Installed to the home screen (the manifest and a service+worker that caches the shell), it is the phone app: it opens offline and+polls when it can. There is no native app and no app store; the design+does not want a build step between a person and their swarm.++## 3. Keeping++A *blob* is a JSON value under a key, at most 256 KB. A hub keeps its+own blobs (`--store DIR`, `own/`) and keeps blobs for others (`kept/`,+at most 200 keys and 64 MB per actor). Three activities:++- **`Keep`** `{key, sha256, size, updated, data}`: "hold this for me."+ A hub keeper stores it and confirms in the same round trip; a member+ stores it in IndexedDB on its next poll and posts **`Kept`** `{key,+ sha256}` to confirm.+- **`Fetch`** `{key}`: "hand it back." A hub keeper answers with the+ blob at once; a member answers on its next poll with a `Kept` carrying+ `data`, which the hub adopts if it has no copy.+- Every `Announce` from a keeper carries `kept`, the keys it holds for+ the receiver. A hub that finds a key it no longer has sends `Fetch`.+ This is how a box that died and came back with the same name gets its+ blobs without anyone doing anything: its members' next announces name+ what they kept.++**Replication.** `--replicas N` (default 2) is how many fresh holders+each of the hub's own blobs should have; a holder is fresh for a day+after confirming the current version. On every poll the hub sends `Keep`+for each blob short of that, to keepers that do not hold it, members+first, someone new before a holder gone stale; a member asked within+ten minutes counts as covered until it confirms, so one blob does not go+to every phone at once. A new version of a blob starts the count over.++**What is kept automatically:** the hub's directory, its collected gaps+and its unanswered sentences, each re-kept when it changes. **What is+never kept:** the state file. It holds the pair secrets, and a blob+goes to other people's devices.++`pyrtan keep KEY FILE` gives a hub a blob (a recording, a list,+anything JSON); `pyrtan kept` shows what it keeps and who holds each;+`pyrtan fetch KEY` gets one back, from the hub or from whoever holds+it. The operator paths are behind the token; the keeper paths+(`/mailbox`, `/kept/<key>`) are signed.++## 4. What this design refuses++- **A member that can be reached.** No WebRTC, no push service, no+ relay that keeps a socket open. A member is reached by being polled,+ and a hub that wants something from a member waits for its next poll.+ That is the honest shape of a device that sleeps.+- **Secrets as blobs.** Nothing that could let a keeper become the hub.+- **Encryption of blobs.** A blob goes to a keeper that follows the hub+ or is followed by it, over TLS, and holds what the hub's own+ `/directory`, `/gaps` and `/unanswered` already show the hub's+ operator. A private blob given to `pyrtan keep` is the operator's+ call; the standard library has no authenticated encryption, and+ HMAC-CTR of my own is not something to hand people's data to. If+ private blobs matter, that is where a dependency would be justified,+ and the design note should say so when it is taken.+- **Members running trees** in this version. A member announces+ `trees: []`. Routing to a member through its mailbox (`Ask` in,+ `Answer` out, the hub holding the request open) is the next design,+ and needs the JavaScript runtime first.+- **A quorum.** One hub writes each of its blobs; the holders are+ copies, not voters. Last version wins, by the hub's own say.++## 5. Failure modes, honestly++- A hub with no members and no followers keeps its blobs for nobody.+ `pyrtan kept` says "held by 0 of 2" so the operator knows.+- A member that never polls holds nothing new; after a day it is not a+ fresh holder and another is asked. Its mailbox fills to two hundred+ activities and drops the oldest, which for `Keep`s means the hub asks+ again later.+- A hub whose state file is lost loses the pair secrets; its members+ cannot sign to it any more. They re-join from the Member tab (one+ tap), and their next announce names what they kept.+- A blob changed on the hub between a member's `Keep` and its `Kept` is+ confirmed at the old version and re-sent; nothing stale is counted.+- A phone's storage can be cleared by the browser under pressure. The+ replica count is what defends against it: two holders, and the hub's+ own copy.++## 6. Sequencing++- **v0.5, this:** members, mailboxes, keeping, the console as the member+ and the phone app.+- **v0.6:** `pyrtan.js`, the tree runtime in JavaScript, with the Python+ fixtures as shared conformance vectors; then `Ask`/`Answer` through+ the mailbox so a phone answers sentences for its hub.+- **Later:** encrypted private blobs, when a dependency for it is+ justified; members that keep for each other through a hub.diff --git a/deploy/openmediavault/docker-compose.yml b/deploy/openmediavault/docker-compose.yml index 3c003a4..d960924 100644 --- a/deploy/openmediavault/docker-compose.yml +++ b/deploy/openmediavault/docker-compose.yml@@ -40,6 +40,8 @@ x-pyrtan: &pyrtan# PYRTAN_FOLLOW: "seed" # PYRTAN_LISTED: "true" PYRTAN_STATE: /state/federation.json+ PYRTAN_STORE: /state/store # blobs: the hub's own and the ones it keeps for others+ PYRTAN_REPLICAS: "2" # how many members or hubs should keep each of its blobsservices: hub: diff --git a/pyproject.toml b/pyproject.toml index 7f1eef9..e0662ee 100644 --- a/pyproject.toml +++ b/pyproject.toml@@ -1,6 +1,6 @@[project] name = "pyrtan"-version = "0.4.0"+version = "0.5.0"description = "Open-source total automation: a core set of expandable services that act on all forms of I/O through hardcoded, inspectable decision trees." readme = "README.md" license = "AGPL-3.0-or-later" diff --git a/site/README.md b/site/README.md index aabecff..a4c237e 100644 --- a/site/README.md +++ b/site/README.md@@ -7,7 +7,7 @@ static directory and publishes it.| Path | What | Built from | | :-- | :-- | :-- | | `/` | the landing page, with the live directory read from the seeds in the visitor's browser | `site/index.html` |-| `/console/` | the console; point it at any hub started with `--cors https://pyrtan.com` | `pyrtan console --out` |+| `/console/` | the console (installable: manifest and service worker beside it); point it at any hub started with `--cors https://pyrtan.com`; its Member tab joins that hub | `pyrtan console --out` || `/catalog/` | the trees people can `pyrtan pull`, and `index.json` the command reads | `pyrtan index examples`, and the tree files | | `/rendezvous/seeds.json` | hubs to follow first | `site/rendezvous/seeds.json` | | `/rendezvous/directory.json` | a snapshot of the directory as the seeds know it, refreshed hourly | `site/snapshot.py` | diff --git a/src/pyrtan/__init__.py b/src/pyrtan/__init__.py index 6bb53ba..939c739 100644 --- a/src/pyrtan/__init__.py +++ b/src/pyrtan/__init__.py@@ -33,4 +33,4 @@ __all__ = ["Response", ]-__version__ = "0.4.0"+__version__ = "0.5.0"diff --git a/src/pyrtan/catalog.py b/src/pyrtan/catalog.py index 3285d89..aea85b9 100644 --- a/src/pyrtan/catalog.py +++ b/src/pyrtan/catalog.py@@ -148,7 +148,7 @@ def _fetch(url: str, timeout: float = 20.0) -> bytes:raise CatalogError(f"{url}: {e}") from e if parts.scheme not in ("http", "https"): raise CatalogError(f"{url}: a catalog is reached over https (or a file: path)")- req = urllib.request.Request(url, headers={"User-Agent": "pyrtan-catalog/0.4", "Accept": "application/json, application/toml, text/plain, */*"})+ req = urllib.request.Request(url, headers={"User-Agent": "pyrtan-catalog/0.5", "Accept": "application/json, application/toml, text/plain, */*"})try: with urllib.request.urlopen(req, timeout=timeout) as resp: return resp.read() diff --git a/src/pyrtan/cli.py b/src/pyrtan/cli.py index 7768748..a3efed5 100644 --- a/src/pyrtan/cli.py +++ b/src/pyrtan/cli.py@@ -14,6 +14,9 @@pyrtan unfollow URL --at HUB pyrtan peers --at HUB services, and hubs followed and following pyrtan directory [--at HUB] every hub that asked to be found, as the swarm knows it+ pyrtan keep KEY [FILE] --at HUB give your hub a blob to keep, and have its keepers keep it+ pyrtan kept --at HUB what your hub keeps, who holds it, and what it keeps for others+ pyrtan fetch KEY --at HUB get a blob back, from your hub or from whoever holds itpyrtan gaps --at URL branches waiting to be written pyrtan unanswered --at URL [--scaffold DIR] trees waiting to be written pyrtan new [NAME] [--from TEXT] [--dir PATH] scaffold a tree file@@ -41,7 +44,7 @@ from . import __version__from .catalog import Catalog, CatalogError, build_index, catalog_url from .federation import bootstrap_candidates, first_live_seed from .grow import Proposal, propose-from .page import PAGE+from .page import MANIFEST, PAGE, SERVICE_WORKERfrom .registry import Registry, default_paths from .runtime import Outcome, Runner from .service import RemoteResolver, http_json, outcome_from_dict, serve@@ -121,6 +124,8 @@ def main(argv: list[str] | None = None) -> int:p.add_argument("--closed", action="store_true", help="take no follows at all (this hub can still follow others)") p.add_argument("--state", metavar="FILE", default=os.environ.get("PYRTAN_STATE"), help="keep who this hub follows and is followed by, and the directory, in this file across restarts (default $PYRTAN_STATE)") p.add_argument("--listed", action="store_true", default=os.environ.get("PYRTAN_LISTED", "").lower() in ("1", "true", "yes"), help="put this hub in the directory it gossips, so others can find it (default $PYRTAN_LISTED)")+ p.add_argument("--store", metavar="DIR", default=os.environ.get("PYRTAN_STORE"), help="where this hub keeps blobs, its own and others' (default $PYRTAN_STORE, else next to --state)")+ p.add_argument("--replicas", type=int, default=int(os.environ.get("PYRTAN_REPLICAS", "2")), help="how many members or hubs should keep each of this hub's blobs (default $PYRTAN_REPLICAS or 2)")p.add_argument("--quiet", action="store_true") p = sub.add_parser("follow", help="follow another hub, through a running federated hub of yours")@@ -134,6 +139,23 @@ def main(argv: list[str] | None = None) -> int:p.add_argument("--at", default=os.environ.get("PYRTAN_AT", "http://127.0.0.1:8700"), help="your hub (default $PYRTAN_AT or the local hub)") _token_arg(p, "your hub's token, if any")+ p = sub.add_parser("keep", help="give your hub a blob to keep; its keepers get a copy")+ p.add_argument("key", help="a key: lowercase, digits, dot, dash, underscore, slash")+ p.add_argument("file", nargs="?", help="a JSON file to keep (default: read JSON from stdin)")+ p.add_argument("--at", default=os.environ.get("PYRTAN_AT", "http://127.0.0.1:8700"), help="your hub (default $PYRTAN_AT or the local hub)")+ _token_arg(p, "your hub's token, if any")++ p = sub.add_parser("kept", help="what your hub keeps, who holds each blob, and what it keeps for others")+ p.add_argument("--at", default=os.environ.get("PYRTAN_AT", "http://127.0.0.1:8700"))+ p.add_argument("--json", action="store_true")+ _token_arg(p, "your hub's token, if any")++ p = sub.add_parser("fetch", help="get a blob back from your hub, or from whoever holds it")+ p.add_argument("key")+ p.add_argument("--at", default=os.environ.get("PYRTAN_AT", "http://127.0.0.1:8700"))+ p.add_argument("--out", help="write the data here (default: stdout)")+ _token_arg(p, "your hub's token, if any")+p = sub.add_parser("directory", help="every hub that asked to be found, as a hub knows it") p.add_argument("--at", default=None, help="a federated hub to ask (default $PYRTAN_AT, else the local hub, else the first live seed)") p.add_argument("--json", action="store_true")@@ -183,6 +205,7 @@ def main(argv: list[str] | None = None) -> int:"check": cmd_check, "list": cmd_list, "run": cmd_run, "match": cmd_match, "ask": cmd_ask, "serve": cmd_serve, "hub": cmd_hub, "gaps": cmd_gaps, "unanswered": cmd_unanswered, "new": cmd_new, "search": cmd_search, "pull": cmd_pull, "index": cmd_index, "console": cmd_console, "follow": cmd_follow, "unfollow": cmd_unfollow, "peers": cmd_peers, "directory": cmd_directory,+ "keep": cmd_keep, "kept": cmd_kept, "fetch": cmd_fetch,} return commands[args.command](args)@@ -420,6 +443,7 @@ def cmd_hub(args: argparse.Namespace) -> int:federate=args.federate or bool(args.advertise) or bool(follow) or args.listed, advertise=args.advertise, federation_secret=args.federation_secret, state=args.state, open_follows=not args.closed, follow=follow, listed=args.listed, trees=[Path(t) for t in args.trees], env=_kv(args.env), gaps=args.gaps, egress=args.egress,+ store=args.store, replicas=args.replicas,) fed = f", federated as {hub.federation.id}" + (f" following {len(hub.federation.following)}" if hub.federation.following else "") + (", listed" if hub.federation.listed else "") if hub.federation else "" local = f", holding {len(hub.local.trees)} tree(s)" if hub.local else ""@@ -471,6 +495,68 @@ def cmd_follow(args: argparse.Namespace) -> int:return _hub_post(args, "/follow", {"url": url, "listed": bool(args.listed)}, lambda b: f"{args.at} now follows {b['following']['url']} ({b['following']['state']}); it routes there once accepted" + (" and is listed in the directory" if b.get("listed") else ""))+def cmd_keep(args: argparse.Namespace) -> int:+ try:+ data = json.loads(Path(args.file).read_text(encoding="utf-8") if args.file else sys.stdin.read())+ except (OSError, json.JSONDecodeError) as e:+ print(f"pyrtan keep: {e}", file=sys.stderr)+ return 2+ return _hub_post(args, "/keep", {"key": args.key, "data": data}, lambda b: f"{args.at} keeps {b['keeping']['key']} ({b['keeping']['size']} bytes); held by {len(b['keeping']['holders'])} of {b['keeping']['wanted']} wanted keeper(s)")+++def cmd_kept(args: argparse.Namespace) -> int:+ at = args.at.rstrip("/")+ try:+ status, body = http_json("GET", f"{at}/keep", token=args.token)+ except Exception as e:+ print(f"pyrtan: {at}: {e}", file=sys.stderr)+ return 3+ if status == 401:+ print(f"pyrtan: {at} needs a token: pass --token or set $PYRTAN_TOKEN", file=sys.stderr)+ return 3+ if status != 200 or not isinstance(body, dict):+ print(f"pyrtan: {at} is not a federated hub ({status})", file=sys.stderr)+ return 3+ if args.json:+ print(json.dumps(body, indent=2, ensure_ascii=False))+ return 0+ print(f"{at} keeps {len(body['keeping'])} blob(s) of its own, wants {body['replicas']} keeper(s) each; keeps {len(body['kept'])} for {body['store']['actors']} other(s)")+ for k in body["keeping"]:+ print(f" {k['key']:24} {k['size']:>8} B held by {len(k['holders'])}/{k['wanted']}" + (f" ({', '.join(k['holders'])})" if k["holders"] else " (nobody yet)"))+ for k in body["kept"]:+ print(f" for {k['actor']}: {k['key']} ({k['size']} B)")+ return 0+++def cmd_fetch(args: argparse.Namespace) -> int:+ at = args.at.rstrip("/")+ try:+ status, body = http_json("POST", f"{at}/fetch", {"key": args.key}, token=args.token)+ except Exception as e:+ print(f"pyrtan: {at}: {e}", file=sys.stderr)+ return 3+ if status == 401:+ print(f"pyrtan: {at} needs a token: pass --token or set $PYRTAN_TOKEN", file=sys.stderr)+ return 3+ if status != 200 or not isinstance(body, dict):+ print(f"pyrtan: {at} answered {status}", file=sys.stderr)+ return 3+ if body.get("status") == "asked":+ print(f"pyrtan fetch: {body.get('say')}", file=sys.stderr)+ return 4+ if body.get("status") != "ok":+ print(f"pyrtan fetch: {body.get('say') or body.get('reason')}", file=sys.stderr)+ return 1+ blob = body["blob"]+ text = json.dumps(blob["data"], indent=2, ensure_ascii=False)+ if args.out:+ Path(args.out).write_text(text + "\n", encoding="utf-8")+ print(f"{blob['key']} ({blob['size']} bytes, from {body['source']}) -> {args.out}")+ else:+ print(text)+ return 0++def cmd_directory(args: argparse.Namespace) -> int: tried: list[str] = [] candidates = [args.at] if args.at else [os.environ.get("PYRTAN_AT") or "http://127.0.0.1:8700", *bootstrap_candidates()]@@ -651,7 +737,9 @@ def cmd_console(args: argparse.Namespace) -> int:out = Path(args.out) out.parent.mkdir(parents=True, exist_ok=True) out.write_text(PAGE, encoding="utf-8")- print(f"wrote {out}; serve it from any static host, and point it at a hub with --cors set for that host")+ (out.parent / "manifest.webmanifest").write_text(MANIFEST, encoding="utf-8")+ (out.parent / "sw.js").write_text(SERVICE_WORKER, encoding="utf-8")+ print(f"wrote {out}, manifest.webmanifest and sw.js beside it; serve the folder from any static host, and point it at a hub with --cors set for that host")return 0 diff --git a/src/pyrtan/federation.py b/src/pyrtan/federation.py index 759d376..77d1d85 100644 --- a/src/pyrtan/federation.py +++ b/src/pyrtan/federation.py@@ -71,6 +71,7 @@ from typing import Any, Callablefrom . import __version__ from .service import http_json+from .store import BlobStore, StoreError, check_key, digest as blob_digestDEFAULT_SEEDS = "https://pyrtan.com/rendezvous/seeds.json" DEFAULT_SNAPSHOT = "https://pyrtan.com/rendezvous/directory.json"@@ -83,7 +84,11 @@ DIRECTORY_TTL = 7 * 86400 # an entry unheard of for this long is droppedDIRECTORY_MAX = 500 HEARD_UP = 600.0 # a directory entry seen this recently reads "up" HEARD_STALE = 86400.0-ACTIVITY_TYPES = ("Follow", "Accept", "Reject", "Undo", "Announce")+ACTIVITY_TYPES = ("Follow", "Accept", "Reject", "Undo", "Announce", "Keep", "Kept", "Fetch")+MEMBER_PREFIX = "urn:pyrtan:member:" # an actor with no inbox: a browser or a phone, which polls a mailbox+MAILBOX_MAX = 200 # activities a hub holds for a member that has not polled+HOLDER_FRESH = 86400.0 # a holder confirmed within this counts toward the replicas+ASK_WINDOW = 600.0 # a member asked to keep within this counts as covered until it confirmsSIGNED_PATHS = ("/inbox", "/run", "/intent", "/resume", "/capabilities")@@ -206,7 +211,7 @@ class Remote:last_ok: float = 0.0 last_error: str | None = None since: float = field(default_factory=time.time)- kind: str = "hub"+ kind: str = "hub" # hub | member@property def link(self) -> str:@@ -222,14 +227,14 @@ class Remote:return "down" def to_dict(self) -> dict[str, Any]:- return {"url": self.url, "name": self.name, "kind": "hub", "state": self.state, "link": self.link, "trees": [t["name"] for t in self.trees], "last_ok": self.last_ok, "last_error": self.last_error}+ 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}def to_state(self) -> dict[str, Any]:- return {"url": self.url, "inbox": self.inbox, "secret": self.secret, "name": self.name, "state": self.state, "since": self.since}+ return {"url": self.url, "inbox": self.inbox, "secret": self.secret, "name": self.name, "state": self.state, "since": self.since, "kind": self.kind}@classmethod def from_state(cls, d: dict[str, Any]) -> "Remote":- return cls(d["url"], d["inbox"], d["secret"], d.get("name", ""), d.get("state", "pending"), since=float(d.get("since", time.time())))+ 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"))def heard(seen: float, now: float | None = None) -> str:@@ -239,7 +244,7 @@ def heard(seen: float, now: float | None = None) -> str:class Federation:- 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) -> None:+ 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) -> None:self.id = hub_id.rstrip("/") self.name = name self.secret = secret or None # a shared federation secret: Follows must be signed with it@@ -247,6 +252,12 @@ class Federation:self.state_path = Path(state) if state else None self.capabilities = capabilities or (lambda: []) self.listed = listed # put ourselves in the directory we gossip+ self.store = store if isinstance(store, BlobStore) else BlobStore(store)+ self.replicas = max(0, int(replicas)) # how many keepers each of our blobs should have+ self.auto_keys = dict(auto_keys or {}) # blobs kept automatically: key -> what to keep+ self._auto_digest: dict[str, str] = {}+ self.keeping: dict[str, dict[str, Any]] = {} # our blobs: key -> {sha256, size, updated, holders: {actor: {at, sha256}}}+ self.mailboxes: dict[str, deque[dict[str, Any]]] = {} # activities waiting for membersself.following: dict[str, Remote] = {} self.followers: dict[str, Remote] = {} self.directory_entries: dict[str, dict[str, Any]] = {}@@ -267,7 +278,7 @@ class Federation:return f"{self.id}/inbox" def actor(self) -> dict[str, Any]:- return {"type": "Hub", "id": self.id, "name": self.name, "inbox": self.inbox, "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}+ 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}# ------------------------------------------------------------ state@@ -280,6 +291,7 @@ class Federation:"following": [r.to_state() for r in self.following.values()], "followers": [r.to_state() for r in self.followers.values()], "directory": list(self.directory_entries.values()),+ "keeping": self.keeping,} self.state_path.parent.mkdir(parents=True, exist_ok=True) self.state_path.write_text(json.dumps(data, indent=2) + "\n", encoding="utf-8")@@ -296,6 +308,9 @@ class Federation:for d in data.get("followers", []): self.followers[d["url"]] = Remote.from_state(d) self.merge_directory(data.get("directory", []), via="state")+ keeping = data.get("keeping")+ if isinstance(keeping, dict):+ self.keeping = {k: v for k, v in keeping.items() if isinstance(v, dict) and isinstance(v.get("holders"), dict)}def set_listed(self, listed: bool) -> None: with self._lock:@@ -309,7 +324,7 @@ class Federation:def own_entry(self) -> dict[str, Any] | None: if not self.listed: return None- return {"id": self.id, "name": self.name, "trees": sorted({t.get("name") for t in self.capabilities() if isinstance(t, dict) and t.get("name")}), "seen": time.time()}+ 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()}def gossip(self) -> list[dict[str, Any]]: """What this hub tells others about the directory: itself, if@@ -331,7 +346,7 @@ class Federation:changed = 0 with self._lock: for e in entries:- if not isinstance(e, dict) or not isinstance(e.get("id"), str) or not e["id"].startswith(("http://", "https://")):+ if not isinstance(e, dict) or not isinstance(e.get("id"), str) or not e["id"].startswith(("http://", "https://", MEMBER_PREFIX)):continue hub_id = e["id"].rstrip("/") if hub_id == self.id:@@ -348,6 +363,7 @@ class Federation:self.directory_entries[hub_id] = { "id": hub_id, "name": str(e.get("name") or "")[:80],+ "kind": "member" if hub_id.startswith(MEMBER_PREFIX) else "hub","trees": [str(t)[:80] for t in e.get("trees", []) if isinstance(t, (str, int))][:200], "seen": seen, "via": via,@@ -366,7 +382,7 @@ class Federation:now = time.time() hubs = [] for e in self.gossip():- hubs.append({"id": e["id"], "name": e["name"], "trees": e["trees"], "seen": e["seen"], "link": heard(e["seen"], now), "via": e.get("via", "self" if e["id"] == self.id else "")})+ hubs.append({"id": e["id"], "name": e["name"], "kind": e.get("kind", "hub"), "trees": e["trees"], "seen": e["seen"], "link": heard(e["seen"], now), "via": e.get("via", "self" if e["id"] == self.id else "")})hubs.sort(key=lambda h: (h["id"] != self.id, {"up": 0, "stale": 1, "down": 2}[h["link"]], h["name"], h["id"])) return {"hub": self.id, "name": self.name, "listed": self.listed, "count": len(hubs), "hubs": hubs}@@ -375,25 +391,50 @@ class Federation:def _activity(self, type_: str, object_: Any, **extra: Any) -> dict[str, Any]: 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}- def _announcement(self) -> dict[str, Any]:- return {"type": "Capabilities", "name": self.name, "trees": self.capabilities(), "directory": self.gossip()}+ def _announcement(self, for_remote: Remote | None = None) -> dict[str, Any]:+ out = {"type": "Capabilities", "name": self.name, "trees": self.capabilities(), "directory": self.gossip()}+ if for_remote is not None:+ out["kept"] = [b["key"] for b in self.store.list_kept(for_remote.url)] # what we keep for them+ return out- def deliver(self, remote: Remote, activity: dict[str, Any], secret: str | None = None) -> tuple[bool, str | None]:+ def deliver(self, remote: Remote, activity: dict[str, Any], secret: str | None = None) -> tuple[bool, str | None, Any]:"""Post an activity to a remote's inbox, signed with the pair secret- (or the one given). Returns (delivered, error)."""+ (or the one given); for a member, leave it in its mailbox. Returns+ (delivered, error, the inbox's answer)."""+ if remote.inbox == "mailbox":+ box = self.mailboxes.setdefault(remote.url, deque(maxlen=MAILBOX_MAX))+ box.append(activity)+ self.outbox.append({**activity, "to": remote.url, "delivered": True, "error": None, "mailbox": True})+ return True, None, Nonepath = "/inbox" headers = sign(secret or remote.secret, self.id, "POST", path, activity) try: status, answer = http_json("POST", remote.inbox, activity, timeout=10.0, headers=headers) except Exception as e: self.outbox.append({**activity, "to": remote.url, "delivered": False, "error": str(e)})- return False, str(e)+ return False, str(e), Noneok = status in (200, 202) self.outbox.append({**activity, "to": remote.url, "delivered": ok, "error": None if ok else f"{status}: {answer}"})- if ok and isinstance(answer, dict) and activity.get("type") == "Announce":- # An inbox answers an Announce with its own gossip: one round trip, both directions.- self.merge_directory(answer.get("directory"), via=remote.url)- return ok, None if ok else f"inbox answered {status}: {answer}"+ if ok and isinstance(answer, dict):+ if activity.get("type") == "Announce":+ # An inbox answers an Announce with its own gossip: one round trip, both directions.+ self.merge_directory(answer.get("directory"), via=remote.url)+ if activity.get("type") == "Keep" and answer.get("kept") == activity["object"].get("key"):+ self.confirm(remote.url, activity["object"]["key"], activity["object"]["sha256"])+ return ok, None if ok else f"inbox answered {status}: {answer}", answer++ def mailbox(self, actor: str) -> list[dict[str, Any]]:+ """Hand a member what has been waiting for it, and forget it."""+ box = self.mailboxes.get(actor.rstrip("/"))+ if not box:+ return []+ out = list(box)+ box.clear()+ remote = self.followers.get(actor.rstrip("/"))+ if remote is not None:+ remote.last_ok = time.time()+ remote.last_error = None+ return outdef follow(self, url: str) -> Remote: """Follow the hub at ``url``: fetch its actor, make a pair secret,@@ -421,7 +462,7 @@ class Federation:# Under a shared federation secret the Follow is signed with it; in # the open case with the pair secret, which proves nothing yet but # keeps every message signed.- ok, error = self.deliver(remote, activity, self.secret or remote.secret)+ ok, error, _ = self.deliver(remote, activity, self.secret or remote.secret)if not ok: remote.last_error = error elif remote.state == "accepted":@@ -452,10 +493,9 @@ class Federation:if not due or not self.followers: return 0 self._announced = (digest, time.time())- activity = self._activity("Announce", announcement)delivered = 0 for remote in list(self.followers.values()):- ok, error = self.deliver(remote, activity)+ ok, error, _ = self.deliver(remote, self._activity("Announce", self._announcement(remote)))if ok: delivered += 1 remote.last_ok = time.time()@@ -475,10 +515,9 @@ class Federation:if not due or not accepted: return 0 self._announced_up = (digest, time.time())- activity = self._activity("Announce", announcement)delivered = 0 for remote in accepted:- ok, _ = self.deliver(remote, activity)+ ok, _, _ = self.deliver(remote, self._activity("Announce", self._announcement(remote)))delivered += ok return delivered@@ -501,8 +540,155 @@ class Federation:def refresh_all(self) -> None: for remote in list(self.following.values()): self.refresh(remote)+ self.auto()self.announce_up() self.announce()+ self.replicate()++ # ------------------------------------------------------------ keeping++ def keep(self, key: str, data: Any) -> dict[str, Any]:+ """Keep a blob of our own and see that enough others keep it too."""+ meta = self.store.put_own(key, data)+ with self._lock:+ entry = self.keeping.setdefault(key, {"holders": {}})+ if entry.get("sha256") != meta["sha256"]:+ entry["asked"] = {}+ entry.update({"sha256": meta["sha256"], "size": meta["size"], "updated": meta["updated"]})+ self._save()+ self.replicate(only=key)+ return self.keeping_view(key)++ def confirm(self, actor: str, key: str, sha256: str) -> None:+ with self._lock:+ entry = self.keeping.get(key)+ if entry is None:+ return+ entry["holders"][actor.rstrip("/")] = {"at": time.time(), "sha256": sha256}+ entry.get("asked", {}).pop(actor.rstrip("/"), None)+ self._save()++ def holders(self, key: str) -> list[str]:+ """Who holds the current version of a blob, confirmed recently."""+ entry = self.keeping.get(key)+ if not entry:+ return []+ now = time.time()+ 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)++ def keeping_view(self, key: str | None = None) -> Any:+ def view(k: str, e: dict[str, Any]) -> dict[str, Any]:+ 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"])}++ with self._lock:+ if key is not None:+ e = self.keeping.get(key)+ return view(key, e) if e else None+ return [view(k, e) for k, e in sorted(self.keeping.items())]++ def keepers(self) -> list[Remote]:+ """Who we may ask to keep a blob: everyone we share a secret with, members first."""+ remotes = [r for r in self.followers.values() if r.state == "accepted"] + [r for r in self.following.values() if r.state == "accepted"]+ remotes.sort(key=lambda r: (r.kind != "member", r.link != "up", r.url))+ return remotes++ def replicate(self, only: str | None = None, limit: int = 20) -> int:+ """Send ``Keep`` for every blob with fewer fresh holders than+ ``replicas``, to keepers that do not hold it, at most ``limit`` at a+ time. Returns how many were sent."""+ sent = 0+ now = time.time()+ for key in ([only] if only else list(self.keeping)):+ entry = self.keeping.get(key)+ blob = self.store.get_own(key)+ if not entry or not blob:+ continue+ asked = entry.setdefault("asked", {})+ # A member confirms later, through its mailbox: count it as covered+ # for a while rather than asking every member at once.+ have = set(self.holders(key)) | {actor for actor, t in asked.items() if now - float(t) < ASK_WINDOW}+ # Someone new before a holder gone stale: the stale one may be gone for good.+ for remote in sorted(self.keepers(), key=lambda r: r.url in entry["holders"]):+ if len(have) >= self.replicas or sent >= limit:+ break+ if remote.url in have:+ continue+ ok, _, _ = self.deliver(remote, self._activity("Keep", {"key": key, "sha256": blob["sha256"], "size": blob["size"], "updated": blob["updated"], "data": blob["data"]}))+ if ok:+ sent += 1+ have.add(remote.url)+ if remote.inbox == "mailbox":+ asked[remote.url] = now+ return sent++ def auto(self) -> None:+ """Keep the automatic blobs (directory, gaps, ...) when they changed."""+ for key, fn in self.auto_keys.items():+ try:+ data = fn()+ except Exception:+ continue+ sha, size = blob_digest(data)+ if size > 256 * 1024 or self._auto_digest.get(key) == sha:+ continue+ self._auto_digest[key] = sha+ try:+ self.keep(key, data)+ except StoreError:+ continue++ def kept_for(self, actor: str, key: str) -> dict[str, Any] | None:+ """A blob we keep for ``actor``, for it to take back."""+ return self.store.get_kept(actor, key)++ def fetch(self, key: str) -> dict[str, Any]:+ """Our own copy of a blob, or the copy a holder keeps: a hub answers+ at once, a member is asked through its mailbox and answers later."""+ own = self.store.get_own(key)+ if own:+ return {"status": "ok", "source": "own", "blob": own}+ asked: list[str] = []+ entry = self.keeping.get(key, {})+ for actor in sorted(entry.get("holders", {})):+ remote = self.followers.get(actor) or self.following.get(actor)+ if remote is None:+ continue+ if remote.inbox == "mailbox":+ self.deliver(remote, self._activity("Fetch", {"key": key}))+ asked.append(actor)+ continue+ path = f"/kept/{key}"+ try:+ status, blob = http_json("GET", f"{remote.url}{path}", timeout=10.0, headers=sign(remote.secret, self.id, "GET", path, None))+ except Exception:+ continue+ if status == 200 and isinstance(blob, dict) and "data" in blob:+ try:+ self.store.put_own(key, blob["data"])+ except StoreError:+ continue+ return {"status": "ok", "source": actor, "blob": self.store.get_own(key)}+ if asked:+ return {"status": "asked", "say": f"{len(asked)} member(s) asked through their mailbox; try again once they have answered.", "asked": asked}+ return {"status": "error", "reason": "not-held", "say": f"Nobody holds {key!r}."}++ def _adopt(self, key: str, blob: Any, who: str) -> bool:+ """Take a blob a keeper hands back, when we have no copy of our own."""+ if not isinstance(blob, dict) or "data" not in blob or self.store.get_own(key):+ return False+ try:+ self.store.put_own(key, blob["data"])+ except StoreError:+ return False+ with self._lock:+ entry = self.keeping.setdefault(key, {"holders": {}})+ meta = self.store.get_own(key) or {}+ entry.update({"sha256": meta.get("sha256"), "size": meta.get("size"), "updated": meta.get("updated")})+ entry["holders"][who] = {"at": time.time(), "sha256": meta.get("sha256")}+ self._save()+ if key == "directory":+ self.merge_directory(blob.get("data"), via=who)+ return Truedef _take_capabilities(self, remote: Remote, caps: dict[str, Any]) -> None: with self._lock:@@ -545,18 +731,19 @@ class Federation:self.inbox_log.append({**activity, "received": time.time()}) if kind == "Follow": return self._on_follow(actor, activity, h)- if kind == "Announce":- # From a hub we follow: its trees are ours to route to. From a- # follower: only the directory it carries.- followed = self.following.get(actor)- follower = self.followers.get(actor)+ followed = self.following.get(actor)+ follower = self.followers.get(actor)+ if kind in ("Announce", "Keep", "Kept", "Fetch"):+ # Announce from a hub we follow: its trees are ours to route to;+ # from a follower: only the directory it carries. Keeping goes+ # both ways between any two that share a secret.remote = followed or follower else:- remote = self.following.get(actor) if kind in ("Accept", "Reject") else self.followers.get(actor)+ remote = followed if kind in ("Accept", "Reject") else followerif remote is None: return 403, {"status": "error", "reason": "unknown-actor", "say": f"{actor} is not a hub this one follows or is followed by."} problem = verify(remote.secret, "POST", "/inbox", activity, h)- if problem and kind == "Announce" and followed and follower and followed is not follower:+ if problem and kind in ("Announce", "Keep", "Kept", "Fetch") and followed and follower and followed is not follower:# Mutual follow: the announce may be signed with the other pair. remote = follower if remote is followed else followed problem = verify(remote.secret, "POST", "/inbox", activity, h)@@ -587,10 +774,43 @@ class Federation:remote.last_ok = time.time() remote.last_error = None taken = self.merge_directory(obj.get("directory", []), via=actor)- return 200, {"status": "ok", "trees": len(remote.trees) if remote is followed else 0, "merged": taken, "directory": self.gossip()}+ # What it keeps for us that we no longer have: ask for it back.+ wanted = [k for k in obj.get("kept", []) if isinstance(k, str) and not self.store.get_own(k)]+ for k in wanted[:20]:+ self.deliver(remote, self._activity("Fetch", {"key": k}))+ return 200, {"status": "ok", "trees": len(remote.trees) if remote is followed else 0, "merged": taken, "directory": self.gossip(), "kept": [b["key"] for b in self.store.list_kept(actor)]}+ if kind == "Keep":+ obj = activity.get("object")+ if not isinstance(obj, dict) or "data" not in obj:+ return 400, {"status": "error", "reason": "bad-activity", "say": "A Keep carries {\"key\", \"sha256\", \"data\"}."}+ try:+ meta = self.store.put_kept(actor, obj.get("key"), obj["data"], sha256=obj.get("sha256"), updated=obj.get("updated"))+ except StoreError as e:+ return 413, {"status": "error", "reason": "not-kept", "say": str(e)}+ remote.last_ok = time.time()+ return 200, {"status": "ok", "kept": meta["key"], "sha256": meta["sha256"]}+ if kind == "Kept":+ obj = activity.get("object")+ if not isinstance(obj, dict) or not isinstance(obj.get("key"), str):+ return 400, {"status": "error", "reason": "bad-activity", "say": "A Kept names a key."}+ if "data" in obj:+ self._adopt(obj["key"], obj, actor)+ elif isinstance(obj.get("sha256"), str):+ self.confirm(actor, obj["key"], obj["sha256"])+ remote.last_ok = time.time()+ return 200, {"status": "ok"}+ if kind == "Fetch":+ obj = activity.get("object")+ key = obj.get("key") if isinstance(obj, dict) else None+ blob = self.store.get_kept(actor, key) if isinstance(key, str) else None+ if blob is None:+ return 404, {"status": "error", "reason": "not-kept", "say": f"Nothing kept here for {actor} under {key!r}."}+ return 200, {"status": "ok", "blob": blob}if kind == "Undo": with self._lock: self.followers.pop(actor, None)+ self.mailboxes.pop(actor, None)+ self.store.drop_kept(actor)self._save() return 200, {"status": "ok"} return 400, {"status": "error", "reason": "bad-activity"}@@ -610,23 +830,33 @@ class Federation:problem = verify(pair, "POST", "/inbox", activity, h) if problem: return 401, {"status": "error", "reason": "unsigned", "say": problem}- # The inbox must be the actor's own: ask the actor.- try:- status, doc = http_json("GET", f"{actor}/actor", timeout=10.0)- except Exception as e:- return 400, {"status": "error", "reason": "no-actor", "say": f"{actor}/actor did not answer: {e}"}- if status != 200 or not isinstance(doc, dict) or doc.get("id", "").rstrip("/") != actor or doc.get("inbox") != inbox:- return 400, {"status": "error", "reason": "no-actor", "say": f"{actor}/actor does not describe this follower."}+ if inbox == "mailbox":+ # A member: a browser or a phone, with no address of its own. It+ # polls its mailbox here, signed, instead of being posted to.+ if not actor.startswith(MEMBER_PREFIX) or len(actor) < len(MEMBER_PREFIX) + 16:+ return 400, {"status": "error", "reason": "bad-activity", "say": f"A member's id is {MEMBER_PREFIX}<at least 16 characters>."}+ name = str(activity.get("name") or "")[:80]+ kind = "member"+ else:+ # The inbox must be the actor's own: ask the actor.+ try:+ status, doc = http_json("GET", f"{actor}/actor", timeout=10.0)+ except Exception as e:+ return 400, {"status": "error", "reason": "no-actor", "say": f"{actor}/actor did not answer: {e}"}+ if status != 200 or not isinstance(doc, dict) or doc.get("id", "").rstrip("/") != actor or doc.get("inbox") != inbox:+ return 400, {"status": "error", "reason": "no-actor", "say": f"{actor}/actor does not describe this follower."}+ name = str(activity.get("name") or doc.get("name") or "")+ kind = "hub"with self._lock:- remote = Remote(actor, inbox, pair, str(activity.get("name") or doc.get("name") or ""), state="accepted", last_ok=time.time())+ remote = Remote(actor, inbox, pair, name, state="accepted", last_ok=time.time(), kind=kind)self.followers[actor] = remote self._save() self.deliver(remote, self._activity("Accept", {"type": "Follow", "actor": actor, "object": self.id}))- self.deliver(remote, self._activity("Announce", self._announcement()))- return 202, {"status": "ok", "say": f"{self.id} accepts {actor}."}+ self.deliver(remote, self._activity("Announce", self._announcement(remote)))+ return 202, {"status": "ok", "say": f"{self.id} accepts {actor}.", "mailbox": f"{self.id}/mailbox" if kind == "member" else None}# ------------------------------------------------------------ views def to_dict(self) -> dict[str, Any]: with self._lock:- 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(self.directory_entries) + (1 if self.listed else 0)}+ 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(self.directory_entries) + (1 if self.listed else 0), "keeping": len(self.keeping), "store": self.store.usage(), "replicas": self.replicas}diff --git a/src/pyrtan/hub.py b/src/pyrtan/hub.py index 0508fe7..ae91a69 100644 --- a/src/pyrtan/hub.py +++ b/src/pyrtan/hub.py@@ -44,10 +44,11 @@ from . import __version__from .federation import MAX_HOPS, Federation, Remote, sign from .grow import Unanswered, no_match_outcome from .intent import Trigger, read-from .page import PAGE+from .page import MANIFEST, PAGE, SERVICE_WORKERfrom .registry import Registry from .runtime import Outcome-from .service import Server, Service, bearer_ok, http_json, outcome_from_dict+from .service import Server, Service, bearer_ok, http_json, outcome_from_dict, signed_path+from .store import StoreError, check_keySTALE_AFTER = 30.0 DOWN_AFTER = 90.0@@ -133,10 +134,16 @@ class Hub:# ------------------------------------------------------------ federation- def federate(self, advertise: str, *, secret: str | None = None, state: str | None = None, open_follows: bool = True, listed: bool = False) -> Federation:+ def federate(self, advertise: str, *, secret: str | None = None, state: str | None = None, open_follows: bool = True, listed: bool = False, store: str | None = None, replicas: int = 2) -> Federation:"""Give this hub an identity other hubs can follow. ``advertise`` is- the URL they reach it at."""- self.federation = Federation(advertise, name=self.name, secret=secret, state=state, open_follows=open_follows, capabilities=self.local_capabilities, listed=listed)+ the URL they reach it at. ``store`` is where it keeps blobs, its own+ and others'; ``replicas`` how many keepers each of its own should have."""+ auto = {+ "directory": lambda: self.federation.gossip() if self.federation else [],+ "gaps": lambda: self.gaps()["gaps"],+ "unanswered": lambda: self.unanswered_all()["unanswered"],+ }+ self.federation = Federation(advertise, name=self.name, secret=secret, state=state, open_follows=open_follows, capabilities=self.local_capabilities, listed=listed, store=store, replicas=replicas, auto_keys=auto)return self.federation def hold(self, service: Service) -> Local:@@ -377,10 +384,16 @@ class Hub:if signed and not bearer_ok(headers.get("Authorization"), self.token or ""): # A hub we federate with is asking. It may ask the routed paths # and the inbox; anything else needs the operator's token.- if fed is None or path not in ("/inbox", "/run", "/intent", "/resume", "/capabilities"):- return 401, {"status": "error", "reason": "unauthorised", "say": "A signed request may only reach the inbox and the routed paths."}- if path != "/inbox" and fed.verify_request(method, path, body if method == "POST" else None, headers) is None:+ if fed is None or not signed_path(path):+ return 401, {"status": "error", "reason": "unauthorised", "say": "A signed request may only reach the inbox, the mailbox, the kept blobs and the routed paths."}+ signer = fed.verify_request(method, path, body if method == "POST" else None, headers) if path != "/inbox" else None+ if path != "/inbox" and signer is None:return 401, {"status": "error", "reason": "unsigned", "say": "The signature does not hold, or this hub does not know the actor."}+ if method == "GET" and path == "/mailbox":+ return 200, {"activities": fed.mailbox(signer or "")}+ if method == "GET" and path.startswith("/kept/"):+ blob = fed.kept_for(signer or "", path[len("/kept/"):])+ return (200, blob) if blob else (404, {"status": "error", "reason": "not-kept", "say": f"Nothing kept here for you under {path[len('/kept/'):]!r}."})if method == "GET" and path == "/health": return 200, self.health() if method == "GET" and path == "/peers":@@ -417,6 +430,21 @@ class Hub:fed.announce_up(force=True) fed.announce(force=True) return 200, {"status": "ok", "listed": fed.listed}+ if method == "GET" and path == "/keep":+ return 200, {"keeping": fed.keeping_view(), "kept": fed.store.list_kept(), "store": fed.store.usage(), "replicas": fed.replicas}+ if method == "GET" and path.startswith("/keep/"):+ blob = fed.store.get_own(path[len("/keep/"):])+ return (200, {**blob, "holders": fed.holders(blob["key"])}) if blob else (404, {"status": "error", "reason": "no-blob", "say": f"No blob of ours under {path[len('/keep/'):]!r}; POST /fetch asks the holders."})+ if method == "POST" and path == "/keep":+ try:+ return 200, {"status": "ok", "keeping": fed.keep(check_key(body.get("key")), body.get("data"))}+ except StoreError as e:+ return 400, {"status": "error", "reason": "bad-blob", "say": str(e)}+ if method == "POST" and path == "/fetch":+ key = body.get("key")+ if not isinstance(key, str):+ return 400, {"status": "error", "reason": "bad-request", "say": "Send {\"key\": \"...\"}."}+ return 200, fed.fetch(key)if method == "POST" and path == "/unfollow": url = body.get("url") if not isinstance(url, str):@@ -430,10 +458,14 @@ class Hub:return 200, self.unanswered_all() if method == "GET" and path in ("/", "/index.html", "/console"): return 200, PAGE+ if method == "GET" and path == "/manifest.webmanifest":+ return 200, ("application/manifest+json", MANIFEST)+ if method == "GET" and path == "/sw.js":+ return 200, ("text/javascript; charset=utf-8", SERVICE_WORKER)if method == "GET" and path == "/api": endpoints = ["/health", "/peers", "/capabilities", "/intent", "/run", "/resume", "/gaps", "/unanswered", "/register"] if fed is not None:- endpoints += ["/actor", "/inbox", "/directory", "/follow", "/unfollow", "/listed", "/following", "/followers", "/outbox"]+ endpoints += ["/actor", "/inbox", "/mailbox", "/directory", "/follow", "/unfollow", "/listed", "/following", "/followers", "/outbox", "/keep", "/fetch", "/kept/{key}"]return 200, {"pyrtan": "hub", "name": self.name, "version": __version__, "auth": "bearer" if self.token else "none", "federation": fed.id if fed else None, "endpoints": endpoints} if method == "POST" and path == "/register": url = body.get("url")@@ -476,12 +508,14 @@ def serve_hub(env: dict[str, Any] | None = None, gaps: str | None = None, egress: str = "any",+ store: str | None = None,+ replicas: int = 2,) -> tuple[Hub, Server]: local = Service(Registry.from_paths(trees), name=f"{name}", env=env, gaps=gaps, egress=egress) if trees else None hub = Hub(name=name, poll_every=poll_every, token=token, local=local) server = Server(hub.route, host, port, quiet, token=token, cors=cors, federated=federate or bool(advertise)).start() if federate or advertise:- hub.federate(advertise or server.url, secret=federation_secret, state=state, open_follows=open_follows, listed=listed)+ hub.federate(advertise or server.url, secret=federation_secret, state=state, open_follows=open_follows, listed=listed, store=store or (str(Path(state).with_name("store")) if state else None), replicas=replicas)for url in peers or []: hub.add_peer(url) for url in follow or []: diff --git a/src/pyrtan/page.py b/src/pyrtan/page.py index 9f0b273..0fdca9a 100644 --- a/src/pyrtan/page.py +++ b/src/pyrtan/page.py@@ -18,6 +18,9 @@ console's origin with ``--cors``. No external assets, no build step.from __future__ import annotations+import json+import urllib.parse+from . import __version__ PAGE = r"""<!doctype html>@@ -26,7 +29,9 @@ PAGE = r"""<!doctype html><meta charset="utf-8"> <meta name="viewport" content="width=device-width, initial-scale=1"> <title>Pyrtan</title>-<meta name="description" content="Pyrtan console: ask the swarm, read the trace, see what it can answer, and find the gaps.">+<meta name="description" content="Pyrtan console: ask the swarm, read the trace, see what it can answer, find the gaps, and join a hub as a member that keeps its blobs.">+<link rel="manifest" href="manifest.webmanifest">+<meta name="theme-color" content="#2b5db8"><style> :root { --bg:#fbfaf7; --fg:#1d1d1b; --mute:#6b6a64; --line:#e3e1da; --card:#ffffff; --ok:#1f7a4d; --bad:#a3382b; --ask:#8a6400; --accent:#2b5db8; --accent-fg:#fff; --soft:#f1efe9 } @media (prefers-color-scheme: dark) { :root { --bg:#141513; --fg:#ecebe6; --mute:#9d9c95; --line:#2b2c29; --card:#1c1d1b; --ok:#5fcf93; --bad:#f08a7a; --ask:#e2b84a; --accent:#7fa6f0; --accent-fg:#101318; --soft:#222420 } }@@ -95,6 +100,7 @@ PAGE = r"""<!doctype html><a href="#gaps" data-tab="gaps">Gaps</a> <a href="#unanswered" data-tab="unanswered">Unanswered</a> <a href="#directory" data-tab="directory" id="directory-tab" hidden>Directory</a>+ <a href="#member" data-tab="member" id="member-tab" hidden>Member</a><a href="#api" data-tab="api">API</a> </nav>@@ -145,6 +151,33 @@ PAGE = r"""<!doctype html><div class="card"><table id="directory"><thead><tr><th>Heard</th><th>Hub</th><th>Holds</th><th>Last announced</th></tr></thead><tbody></tbody></table></div> </section>+ <section id="tab-member" hidden>+ <h2>This browser as a member</h2>+ <p class="mute" style="font-size:14px;margin-top:0">A browser or a phone cannot be reached, so it joins a hub the other way round: it follows the hub, polls a mailbox the hub keeps for it, carries the directory, and keeps copies of the hub's blobs (its directory, gaps, unanswered sentences, anything given to <code>pyrtan keep</code>). When the hub's box dies, what it kept is still here. Install this page to the home screen and it is the phone app.</p>+ <div class="card">+ <div id="member-off">+ <form class="row" id="join-form">+ <input type="text" id="member-name" placeholder="a name for this device, e.g. barry's phone" autocomplete="off">+ <input type="password" id="member-fedsecret" placeholder="federation secret, if the hub takes follows only with one" autocomplete="off">+ <button type="submit">Join this hub</button>+ </form>+ <div class="mute" style="font-size:14px">Makes an identity for this browser (kept in local storage), follows the hub with it, and starts polling. Nothing leaves this device but signed activities to the hub.</div>+ </div>+ <div id="member-on" hidden>+ <div><span class="status done">member</span> <code id="member-id"></code> <span class="mute" id="member-state"></span></div>+ <div class="mute" style="font-size:14px;margin:6px 0">Name <code id="member-name-shown"></code> · mailbox polled <span id="member-polled">never</span> · <label><input type="checkbox" id="member-listed"> listed in the directory</label></div>+ <div style="display:flex;gap:8px;flex-wrap:wrap;margin:8px 0">+ <button type="button" class="quiet" id="member-poll">Poll now</button>+ <button type="button" class="quiet" id="member-leave">Leave</button>+ </div>+ <h2 style="font-size:15px;margin:14px 0 6px">Kept here for this hub</h2>+ <table id="kept"><thead><tr><th>Key</th><th>Size</th><th>Updated</th><th></th></tr></thead><tbody></tbody></table>+ <h2 style="font-size:15px;margin:14px 0 6px">Activity</h2>+ <pre id="member-log" style="max-height:220px"></pre>+ </div>+ </div>+ </section>+<section id="tab-api" hidden> <h2>The same endpoints this page uses</h2> <div class="card">@@ -202,6 +235,7 @@ function showTab(name) {if (name === "gaps") loadGaps(); if (name === "unanswered") loadUnanswered(); if (name === "directory") loadDirectory();+ if (name === "member") member.render();} async function loadDirectory() {@@ -209,7 +243,7 @@ async function loadDirectory() {const body = $("#directory tbody"); body.innerHTML = ""; for (const h of d.hubs || []) { const tr = document.createElement("tr");- tr.innerHTML = `<td><span class="link ${esc(h.link)}"></span>${esc(h.link)}</td><td><code>${esc(h.name||"")}</code>${h.via === "self" ? ' <span class="mute">(this hub)</span>' : ""}<div class="mute">${esc(h.id)}</div></td><td>${(h.trees||[]).map(n => `<code>${esc(n)}</code>`).join(" ")}</td><td class="mute">${h.seen ? new Date(h.seen * 1000).toLocaleString() : ""}</td>`;+ tr.innerHTML = `<td><span class="link ${esc(h.link)}"></span>${esc(h.link)}</td><td><code>${esc(h.name||"")}</code>${h.via === "self" ? ' <span class="mute">(this hub)</span>' : ""}${h.kind === "member" ? ' <span class="mute">(member)</span>' : ""}<div class="mute">${esc(h.id)}</div></td><td>${(h.trees||[]).map(n => `<code>${esc(n)}</code>`).join(" ")}</td><td class="mute">${h.seen ? new Date(h.seen * 1000).toLocaleString() : ""}</td>`;body.appendChild(tr); } if (!(d.hubs || []).length) body.innerHTML = `<tr><td colspan="4" class="mute">Nobody heard of yet. Follow a seed, or start this hub with <code>--listed</code>.</td></tr>`;@@ -264,6 +298,9 @@ async function load() {if (!a) { $("#who").textContent = remote ? "not connected" : "unreachable"; return; } $("#peers-tab").hidden = !isHub; $("#directory-tab").hidden = !(isHub && a.federation);+ $("#member-tab").hidden = !(isHub && a.federation);+ member.hub = a.federation || null;+ if (member.hub) member.start();$("#where").textContent = (base || location.origin) + (a.auth === "bearer" ? " · token" : ""); $("#hub-version").textContent = a.version ? ` · ${a.pyrtan} v${a.version}` : ""; try {@@ -382,9 +419,175 @@ $("#connect-form").addEventListener("submit", (e) => {store.set(remote ? {url: $("#hub-url").value.trim(), token} : {token}); load(); });+// ---------------------------------------------------------------- member+// A browser as a federation actor without an inbox. Everything it sends is+// signed the way hubs sign: HMAC-SHA256 over method, path, date and the+// SHA-256 of the canonical JSON body. SHA-256 and HMAC are written out here+// because crypto.subtle is not available on a plain http:// LAN address.+const sha256 = (() => {+ const K = [0x428a2f98,0x71374491,0xb5c0fbcf,0xe9b5dba5,0x3956c25b,0x59f111f1,0x923f82a4,0xab1c5ed5,0xd807aa98,0x12835b01,0x243185be,0x550c7dc3,0x72be5d74,0x80deb1fe,0x9bdc06a7,0xc19bf174,0xe49b69c1,0xefbe4786,0x0fc19dc6,0x240ca1cc,0x2de92c6f,0x4a7484aa,0x5cb0a9dc,0x76f988da,0x983e5152,0xa831c66d,0xb00327c8,0xbf597fc7,0xc6e00bf3,0xd5a79147,0x06ca6351,0x14292967,0x27b70a85,0x2e1b2138,0x4d2c6dfc,0x53380d13,0x650a7354,0x766a0abb,0x81c2c92e,0x92722c85,0xa2bfe8a1,0xa81a664b,0xc24b8b70,0xc76c51a3,0xd192e819,0xd6990624,0xf40e3585,0x106aa070,0x19a4c116,0x1e376c08,0x2748774c,0x34b0bcb5,0x391c0cb3,0x4ed8aa4a,0x5b9cca4f,0x682e6ff3,0x748f82ee,0x78a5636f,0x84c87814,0x8cc70208,0x90befffa,0xa4506ceb,0xbef9a3f7,0xc67178f2];+ const rotr = (x, n) => (x >>> n) | (x << (32 - n));+ return (bytes) => {+ const l = bytes.length, bitLen = l * 8, padded = new Uint8Array(((l + 9 + 63) >> 6) << 6);+ padded.set(bytes); padded[l] = 0x80;+ const dv = new DataView(padded.buffer); dv.setUint32(padded.length - 4, bitLen >>> 0); dv.setUint32(padded.length - 8, Math.floor(bitLen / 4294967296));+ let h = [0x6a09e667,0xbb67ae85,0x3c6ef372,0xa54ff53a,0x510e527f,0x9b05688c,0x1f83d9ab,0x5be0cd19];+ const w = new Uint32Array(64);+ for (let o = 0; o < padded.length; o += 64) {+ for (let i = 0; i < 16; i++) w[i] = dv.getUint32(o + i * 4);+ for (let i = 16; i < 64; i++) { const s0 = rotr(w[i-15],7) ^ rotr(w[i-15],18) ^ (w[i-15] >>> 3), s1 = rotr(w[i-2],17) ^ rotr(w[i-2],19) ^ (w[i-2] >>> 10); w[i] = (w[i-16] + s0 + w[i-7] + s1) >>> 0; }+ let [a,b,c,d,e,f,g,hh] = h;+ for (let i = 0; i < 64; i++) { const S1 = rotr(e,6) ^ rotr(e,11) ^ rotr(e,25), ch = (e & f) ^ (~e & g), t1 = (hh + S1 + ch + K[i] + w[i]) >>> 0, S0 = rotr(a,2) ^ rotr(a,13) ^ rotr(a,22), maj = (a & b) ^ (a & c) ^ (b & c), t2 = (S0 + maj) >>> 0; hh = g; g = f; f = e; e = (d + t1) >>> 0; d = c; c = b; b = a; a = (t1 + t2) >>> 0; }+ h = h.map((v, i) => (v + [a,b,c,d,e,f,g,hh][i]) >>> 0);+ }+ return new Uint8Array(h.flatMap(v => [v >>> 24 & 255, v >>> 16 & 255, v >>> 8 & 255, v & 255]));+ };+})();+const enc = new TextEncoder();+const hex = (b) => Array.from(b, x => x.toString(16).padStart(2, "0")).join("");+function hmac(keyText, msgBytes) {+ let key = enc.encode(keyText); if (key.length > 64) key = sha256(key);+ const k = new Uint8Array(64); k.set(key);+ const ipad = k.map(x => x ^ 0x36), opad = k.map(x => x ^ 0x5c);+ const inner = sha256(new Uint8Array([...ipad, ...msgBytes]));+ return hex(sha256(new Uint8Array([...opad, ...inner])));+}+function canonical(v) {+ if (v === null || typeof v !== "object") return JSON.stringify(v);+ if (Array.isArray(v)) return "[" + v.map(canonical).join(",") + "]";+ return "{" + Object.keys(v).sort().map(k => JSON.stringify(k) + ":" + canonical(v[k])).join(",") + "}";+}+function signHeaders(secret, actor, method, path, body) {+ const date = String(Math.floor(Date.now() / 1000));+ const digest = hex(sha256(enc.encode(body === null || body === undefined ? "" : canonical(body))));+ const mac = hmac(secret, enc.encode([method.toUpperCase(), path, date, digest].join("\n")));+ return {"X-Pyrtan-Actor": actor, "X-Pyrtan-Date": date, "X-Pyrtan-Signature": mac};+}+const randomHex = (n) => hex(crypto.getRandomValues(new Uint8Array(n)));++const member = {+ hub: null, me: null, timer: null, log: [], directory: new Map(), db: null,+ key() { return "pyrtan.member:" + (this.hub || ""); },+ load() { try { this.me = JSON.parse(localStorage.getItem(this.key()) || "null"); } catch (e) { this.me = null; } },+ save() { try { if (this.me) localStorage.setItem(this.key(), JSON.stringify(this.me)); else localStorage.removeItem(this.key()); } catch (e) {} },+ note(line) { this.log.unshift(new Date().toLocaleTimeString() + " " + line); this.log = this.log.slice(0, 40); const el = $("#member-log"); if (el) el.textContent = this.log.join("\n"); },+ async open() {+ if (this.db) return this.db;+ this.db = await new Promise((res, rej) => { const r = indexedDB.open("pyrtan-member", 1); r.onupgradeneeded = () => r.result.createObjectStore("kept", {keyPath: "id"}); r.onsuccess = () => res(r.result); r.onerror = () => rej(r.error); });+ return this.db;+ },+ async keptAll() { const db = await this.open(); return new Promise((res, rej) => { const r = db.transaction("kept").objectStore("kept").getAll(); r.onsuccess = () => res(r.result.filter(b => b.hub === this.hub)); r.onerror = () => rej(r.error); }); },+ async keptGet(key) { const db = await this.open(); return new Promise((res, rej) => { const r = db.transaction("kept").objectStore("kept").get(this.hub + "|" + key); r.onsuccess = () => res(r.result || null); r.onerror = () => rej(r.error); }); },+ async keptPut(blob) { const db = await this.open(); return new Promise((res, rej) => { const t = db.transaction("kept", "readwrite"); t.objectStore("kept").put({...blob, id: this.hub + "|" + blob.key, hub: this.hub}); t.oncomplete = () => res(); t.onerror = () => rej(t.error); }); },+ async keptClear() { const all = await this.keptAll(); const db = await this.open(); await new Promise((res) => { const t = db.transaction("kept", "readwrite"); for (const b of all) t.objectStore("kept").delete(b.id); t.oncomplete = () => res(); }); },+ async send(method, path, body) {+ if (!this.me) throw new Error("not a member");+ const headers = Object.assign({"Accept": "application/json"}, signHeaders(this.me.secret, this.me.id, method, path, body === undefined ? null : body));+ if (body !== undefined) headers["Content-Type"] = "application/json";+ const r = await fetch((base ? base.replace(/\/+$/, "") : "") + path, {method, headers, body: body === undefined ? undefined : canonical(body)});+ let data = null; try { data = await r.json(); } catch (e) {}+ return {status: r.status, data};+ },+ activity(type, object, extra) { return Object.assign({type, id: this.me.id + "/activities/" + randomHex(8), actor: this.me.id, published: new Date().toISOString().replace(/\.\d+Z$/, "Z"), object}, extra || {}); },+ async join(name, fedSecret) {+ this.me = {id: "urn:pyrtan:member:" + randomHex(16), secret: randomHex(32), name: name || "browser", state: "pending", listed: true, joined: Date.now()};+ this.save();+ const act = this.activity("Follow", this.hub, {secret: this.me.secret, inbox: "mailbox", name: this.me.name});+ const headers = Object.assign({"Accept": "application/json", "Content-Type": "application/json"}, signHeaders(fedSecret || this.me.secret, this.me.id, "POST", "/inbox", act));+ const r = await fetch((base ? base.replace(/\/+$/, "") : "") + "/inbox", {method: "POST", headers, body: canonical(act)});+ const data = await r.json().catch(() => ({}));+ if (r.status !== 202) { this.me = null; this.save(); this.note(`follow refused: ${data.say || r.status}`); problem(`The hub did not take the follow: ${data.say || r.status}`, true); this.render(); return; }+ this.note(`followed ${this.hub}`);+ await this.poll();+ this.start();+ this.render();+ },+ async leave() {+ try { await this.send("POST", "/inbox", this.activity("Undo", {type: "Follow", actor: this.me.id, object: this.hub})); } catch (e) {}+ await this.keptClear();+ this.me = null; this.save(); this.stop(); this.note("left"); this.render();+ },+ async poll() {+ if (!this.me) return;+ let r; try { r = await this.send("GET", "/mailbox"); } catch (e) { this.note("poll failed: " + e.message); return; }+ if (r.status === 401) { this.note("the hub no longer knows this member"); this.me.state = "refused"; this.save(); this.render(); return; }+ this.me.polled = Date.now();+ for (const a of (r.data && r.data.activities) || []) {+ if (a.type === "Accept") { this.me.state = "accepted"; this.note("accepted"); }+ else if (a.type === "Announce") { for (const e of a.object.directory || []) this.directory.set(e.id, e); this.note(`announce: ${(a.object.trees||[]).length} trees, ${(a.object.directory||[]).length} directory entries`); }+ else if (a.type === "Keep") { await this.keptPut(a.object); await this.send("POST", "/inbox", this.activity("Kept", {key: a.object.key, sha256: a.object.sha256})); this.note(`keeping ${a.object.key} (${a.object.size} B)`); }+ else if (a.type === "Fetch") { const b = await this.keptGet(a.object.key); if (b) { await this.send("POST", "/inbox", this.activity("Kept", {key: b.key, sha256: b.sha256, data: b.data})); this.note(`handed back ${b.key}`); } else this.note(`asked for ${a.object.key}, not kept here`); }+ else if (a.type === "Reject") { this.me.state = "rejected"; this.note("rejected: " + (a.summary || "")); }+ }+ this.save(); this.render();+ },+ async announce() {+ if (!this.me || this.me.state !== "accepted") return;+ const kept = (await this.keptAll()).map(b => b.key).sort();+ const own = this.me.listed ? [{id: this.me.id, name: this.me.name, kind: "member", trees: [], seen: Date.now() / 1000}] : [];+ const heard = [...this.directory.values()].filter(e => e.id !== this.me.id);+ try {+ const r = await this.send("POST", "/inbox", this.activity("Announce", {type: "Capabilities", name: this.me.name, trees: [], directory: own.concat(heard), kept}));+ for (const e of (r.data && r.data.directory) || []) this.directory.set(e.id, e);+ } catch (e) { this.note("announce failed: " + e.message); }+ },+ start() {+ this.load();+ if (!this.me || this.timer) { this.render(); return; }+ this.timer = setInterval(() => this.poll(), 20000);+ this.announceTimer = setInterval(() => this.announce(), 60000);+ this.poll().then(() => this.announce());+ },+ stop() { clearInterval(this.timer); clearInterval(this.announceTimer); this.timer = null; },+ async render() {+ if (!$("#member-on")) return;+ $("#member-off").hidden = !!this.me; $("#member-on").hidden = !this.me;+ if (!this.me) return;+ $("#member-id").textContent = this.me.id; $("#member-state").textContent = this.me.state;+ $("#member-name-shown").textContent = this.me.name; $("#member-polled").textContent = this.me.polled ? new Date(this.me.polled).toLocaleTimeString() : "never";+ $("#member-listed").checked = !!this.me.listed;+ const body = $("#kept tbody"); body.innerHTML = "";+ for (const b of await this.keptAll()) { const tr = document.createElement("tr"); tr.innerHTML = `<td><code>${esc(b.key)}</code></td><td>${esc(b.size)} B</td><td class="mute">${b.updated ? new Date(b.updated * 1000).toLocaleString() : ""}</td><td><a href="#" data-show="${esc(b.key)}">show</a></td>`; body.appendChild(tr); }+ if (!body.children.length) body.innerHTML = `<tr><td colspan="4" class="mute">Nothing yet. The hub sends its blobs as it needs keepers; "Poll now" asks sooner.</td></tr>`;+ },+};+$("#join-form").addEventListener("submit", (e) => { e.preventDefault(); member.join($("#member-name").value.trim(), $("#member-fedsecret").value); });+$("#member-poll").addEventListener("click", () => member.poll().then(() => member.announce()));+$("#member-leave").addEventListener("click", () => member.leave());+$("#member-listed").addEventListener("change", (e) => { member.me.listed = e.target.checked; member.save(); member.announce(); });+$("#kept").addEventListener("click", async (e) => { const a = e.target.closest("a[data-show]"); if (!a) return; e.preventDefault(); const b = await member.keptGet(a.dataset.show); alert(b ? JSON.stringify(b.data, null, 2).slice(0, 4000) : "not kept"); });+if ("serviceWorker" in navigator && (location.protocol === "https:" || ["localhost", "127.0.0.1"].includes(location.hostname))) { navigator.serviceWorker.register("sw.js").catch(() => {}); }+$("#forget").addEventListener("click", () => { store.clear(); base = ""; token = ""; $("#hub-url").value = ""; $("#hub-token").value = ""; $("#who").textContent = "not connected"; $("#caps tbody").innerHTML = ""; $("#tries").innerHTML = ""; problem(""); }); load(); </script> </body> </html> """.replace("__VERSION__", __version__)++MANIFEST = json.dumps({+ "name": "Pyrtan",+ "short_name": "Pyrtan",+ "description": "The console of a Pyrtan swarm, and a member of it: install it and this device keeps the hub's blobs.",+ "start_url": "./",+ "scope": "./",+ "display": "standalone",+ "background_color": "#fbfaf7",+ "theme_color": "#2b5db8",+ "icons": [{"src": "data:image/svg+xml," + urllib.parse.quote('<svg xmlns="http://www.w3.org/2000/svg" viewBox="0 0 64 64"><rect width="64" height="64" rx="12" fill="#2b5db8"/><path d="M18 46V18h14a9 9 0 0 1 0 18h-8" fill="none" stroke="#fff" stroke-width="6" stroke-linecap="round" stroke-linejoin="round"/></svg>'), "sizes": "any", "type": "image/svg+xml", "purpose": "any"}],+}, indent=2)++# The service worker keeps the console shell so the page opens offline (a+# phone in a tunnel, a hub that is down): the shell comes from the cache,+# everything the page asks the hub for goes to the network and nowhere else.+SERVICE_WORKER = r"""const SHELL = "pyrtan-shell-__VERSION__";+const FILES = ["./", "manifest.webmanifest"];+self.addEventListener("install", (e) => { e.waitUntil(caches.open(SHELL).then(c => c.addAll(FILES)).then(() => self.skipWaiting())); });+self.addEventListener("activate", (e) => { e.waitUntil(caches.keys().then(keys => Promise.all(keys.filter(k => k !== SHELL).map(k => caches.delete(k)))).then(() => self.clients.claim())); });+self.addEventListener("fetch", (e) => {+ const url = new URL(e.request.url);+ const shell = e.request.mode === "navigate" || url.pathname.endsWith("/manifest.webmanifest");+ if (!shell || e.request.method !== "GET") return; // API calls: the network, always+ e.respondWith(fetch(e.request).then(r => { const copy = r.clone(); caches.open(SHELL).then(c => c.put(e.request, copy)); return r; }).catch(() => caches.match(e.request).then(r => r || caches.match("./"))));+});+""".replace("__VERSION__", __version__)diff --git a/src/pyrtan/service.py b/src/pyrtan/service.py index 6e7de4a..67d6e10 100644 --- a/src/pyrtan/service.py +++ b/src/pyrtan/service.py@@ -53,7 +53,7 @@ from typing import Any, Callablefrom . import __version__ from .gaps import GapStore-from .page import PAGE+from .page import MANIFEST, PAGE, SERVICE_WORKERfrom .grow import Unanswered from .registry import Registry from .runtime import Outcome, Runner@@ -63,10 +63,15 @@ from .transport import Transport, UrllibTransportMAX_BODY = 1 << 20 # What answers without a token: the console, its index, and liveness.-PUBLIC_PATHS = frozenset({"/", "/index.html", "/console", "/api", "/health", "/actor", "/directory"})+PUBLIC_PATHS = frozenset({"/", "/index.html", "/console", "/api", "/health", "/actor", "/directory", "/manifest.webmanifest", "/sw.js"})# On a federated hub, a request carrying a signature reaches the route, # which verifies it (see :mod:`pyrtan.federation`).-SIGNED_PATHS = frozenset({"/inbox", "/run", "/intent", "/resume", "/capabilities"})+SIGNED_PATHS = frozenset({"/inbox", "/run", "/intent", "/resume", "/capabilities", "/mailbox"})+++def signed_path(path: str) -> bool:+ """Whether a signed (federated) request may reach this path at all."""+ return path in SIGNED_PATHS or path.startswith("/kept/")# What /health says to a caller without a token, when one is required. PUBLIC_HEALTH_KEYS = ("status", "name", "peers", "up", "stale", "down", "uptime_s")@@ -223,6 +228,10 @@ class Service:return 200, {"unanswered": self.unanswered.to_list()} if method == "GET" and path in ("/", "/index.html", "/console"): return 200, PAGE+ if method == "GET" and path == "/manifest.webmanifest":+ return 200, ("application/manifest+json", MANIFEST)+ if method == "GET" and path == "/sw.js":+ return 200, ("text/javascript; charset=utf-8", SERVICE_WORKER)if method == "GET" and path == "/api": return 200, {"pyrtan": "service", "name": self.name, "version": __version__, "auth": "bearer" if self.token else "none", "endpoints": ["/health", "/capabilities", "/intent", "/run", "/resume", "/gaps", "/unanswered"]} if method == "POST" and path == "/intent":@@ -287,7 +296,7 @@ def make_handler(route: Route, quiet: bool = False, token: str | None = None, co"""True if the request may proceed; otherwise a 401 has been sent.""" if token is None or path in PUBLIC_PATHS or bearer_ok(self.headers.get("Authorization"), token): return True- if federated and path in SIGNED_PATHS and self.headers.get("X-Pyrtan-Signature"):+ if federated and signed_path(path) and self.headers.get("X-Pyrtan-Signature"):return True # the route verifies the signature self._send(401, {"status": "error", "reason": "unauthorised", "say": "This Pyrtan needs a token: send Authorization: Bearer <token>."}, extra={"WWW-Authenticate": 'Bearer realm="pyrtan"'}) return False@@ -301,7 +310,10 @@ def make_handler(route: Route, quiet: bool = False, token: str | None = None, co# -------------------------------------------------------- i/o def _send(self, status: int, payload: Any, extra: dict[str, str] | None = None) -> None:- if isinstance(payload, str):+ if isinstance(payload, tuple) and len(payload) == 2 and isinstance(payload[0], str):+ ctype, text = payload # (content type, text): the manifest, the service worker+ data = text.encode("utf-8")+ elif isinstance(payload, str):data = payload.encode("utf-8") ctype = "text/html; charset=utf-8" else: diff --git a/src/pyrtan/store.py b/src/pyrtan/store.py new file mode 100644 index 0000000..a5b008e --- /dev/null +++ b/src/pyrtan/store.py@@ -0,0 +1,187 @@+"""Blobs a hub keeps: its own, and the ones it keeps for others.++A *blob* is a JSON value under a key (``directory``, ``gaps``,+``traces/2026-10-25``), with its SHA-256, size and time. A hub keeps its+own blobs so it can hand them to the members and hubs that follow it, and+keeps blobs *for* the hubs and members it shares a secret with, so that+when their box dies the blobs are still somewhere. Nothing secret is ever+a blob: the state file, with the pair secrets, never goes through here.++On disk (``--store DIR``) the layout is ``own/<key>.json`` and+``kept/<actor-hash>/<key>.json``; without a directory the store lives in+memory and is lost with the process, which is enough to try the loop.+Limits are fixed and small on purpose: a blob is at most 256 KB, an+actor may leave at most 200 keys and 64 MB here. A keeper that is asked+for more says so and keeps nothing extra.+"""++from __future__ import annotations++import hashlib+import json+import re+import threading+import time+from pathlib import Path+from typing import Any++MAX_BLOB = 256 * 1024+MAX_KEYS_PER_ACTOR = 200+MAX_BYTES_PER_ACTOR = 64 * 1024 * 1024+KEY = re.compile(r"^[a-z0-9][a-z0-9._/-]{0,99}$")+++class StoreError(ValueError):+ pass+++def canonical(data: Any) -> bytes:+ return json.dumps(data, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode("utf-8")+++def digest(data: Any) -> tuple[str, int]:+ raw = canonical(data)+ return hashlib.sha256(raw).hexdigest(), len(raw)+++def check_key(key: Any) -> str:+ if not isinstance(key, str) or not KEY.match(key) or ".." in key:+ raise StoreError(f"{key!r} is not a key: lowercase letters, digits, dot, dash, underscore, slash; at most 100 characters")+ return key+++def _safe(key: str) -> str:+ return key.replace("/", "__")+++def _actor_dir(actor: str) -> str:+ return hashlib.sha256(actor.rstrip("/").encode("utf-8")).hexdigest()[:16]+++class BlobStore:+ def __init__(self, root: str | Path | None = None) -> None:+ self.root = Path(root) if root else None+ self._own: dict[str, dict[str, Any]] = {} # key -> blob+ self._kept: dict[str, dict[str, dict[str, Any]]] = {} # actor -> key -> blob+ self._lock = threading.RLock()+ if self.root is not None:+ self._load()++ # ------------------------------------------------------------ disk++ def _load(self) -> None:+ assert self.root is not None+ for path in sorted((self.root / "own").glob("*.json")) if (self.root / "own").is_dir() else []:+ blob = self._read(path)+ if blob:+ self._own[blob["key"]] = blob+ kept = self.root / "kept"+ for actor_dir in sorted(kept.iterdir()) if kept.is_dir() else []:+ for path in sorted(actor_dir.glob("*.json")):+ blob = self._read(path)+ if blob and isinstance(blob.get("actor"), str):+ self._kept.setdefault(blob["actor"], {})[blob["key"]] = blob++ @staticmethod+ def _read(path: Path) -> dict[str, Any] | None:+ try:+ blob = json.loads(path.read_text(encoding="utf-8"))+ except (OSError, json.JSONDecodeError):+ return None+ return blob if isinstance(blob, dict) and isinstance(blob.get("key"), str) else None++ def _write(self, path: Path, blob: dict[str, Any]) -> None:+ if self.root is None:+ return+ path.parent.mkdir(parents=True, exist_ok=True)+ tmp = path.with_suffix(".tmp")+ tmp.write_text(json.dumps(blob, ensure_ascii=False) + "\n", encoding="utf-8")+ tmp.replace(path)++ def _remove(self, path: Path) -> None:+ if self.root is not None:+ try:+ path.unlink()+ except OSError:+ pass++ # ------------------------------------------------------------ own++ @staticmethod+ def meta(blob: dict[str, Any]) -> dict[str, Any]:+ return {k: blob[k] for k in ("key", "sha256", "size", "updated") if k in blob}++ def put_own(self, key: str, data: Any) -> dict[str, Any]:+ key = check_key(key)+ sha, size = digest(data)+ if size > MAX_BLOB:+ raise StoreError(f"{key}: {size} bytes; a blob is at most {MAX_BLOB}")+ with self._lock:+ current = self._own.get(key)+ if current is not None and current["sha256"] == sha:+ return self.meta(current)+ blob = {"key": key, "sha256": sha, "size": size, "updated": time.time(), "data": data}+ self._own[key] = blob+ if self.root is not None:+ self._write(self.root / "own" / f"{_safe(key)}.json", blob)+ return self.meta(blob)++ def get_own(self, key: str) -> dict[str, Any] | None:+ with self._lock:+ blob = self._own.get(key)+ return dict(blob) if blob else None++ def list_own(self) -> list[dict[str, Any]]:+ with self._lock:+ return [self.meta(b) for _, b in sorted(self._own.items())]++ # ------------------------------------------------------------ kept for others++ def put_kept(self, actor: str, key: str, data: Any, *, sha256: str | None = None, updated: float | None = None) -> dict[str, Any]:+ key = check_key(key)+ actor = actor.rstrip("/")+ sha, size = digest(data)+ if sha256 is not None and sha256 != sha:+ raise StoreError(f"{key}: the data does not match its sha256")+ if size > MAX_BLOB:+ raise StoreError(f"{key}: {size} bytes; a blob is at most {MAX_BLOB}")+ with self._lock:+ mine = self._kept.setdefault(actor, {})+ others = sum(b["size"] for k, b in mine.items() if k != key)+ if key not in mine and len(mine) >= MAX_KEYS_PER_ACTOR:+ raise StoreError(f"{actor} already has {MAX_KEYS_PER_ACTOR} keys here")+ if others + size > MAX_BYTES_PER_ACTOR:+ raise StoreError(f"{actor} would have more than {MAX_BYTES_PER_ACTOR} bytes here")+ blob = {"actor": actor, "key": key, "sha256": sha, "size": size, "updated": float(updated or time.time()), "kept_at": time.time(), "data": data}+ mine[key] = blob+ if self.root is not None:+ self._write(self.root / "kept" / _actor_dir(actor) / f"{_safe(key)}.json", blob)+ return {**self.meta(blob), "actor": actor}++ def get_kept(self, actor: str, key: str) -> dict[str, Any] | None:+ with self._lock:+ blob = self._kept.get(actor.rstrip("/"), {}).get(key)+ return dict(blob) if blob else None++ def list_kept(self, actor: str | None = None) -> list[dict[str, Any]]:+ with self._lock:+ actors = [actor.rstrip("/")] if actor else sorted(self._kept)+ return [{**self.meta(b), "actor": a} for a in actors for _, b in sorted(self._kept.get(a, {}).items())]++ def drop_kept(self, actor: str) -> int:+ actor = actor.rstrip("/")+ with self._lock:+ mine = self._kept.pop(actor, {})+ for key in mine:+ self._remove((self.root or Path(".")) / "kept" / _actor_dir(actor) / f"{_safe(key)}.json") if self.root else None+ return len(mine)++ def usage(self) -> dict[str, Any]:+ with self._lock:+ return {+ "own": len(self._own),+ "kept": sum(len(m) for m in self._kept.values()),+ "kept_bytes": sum(b["size"] for m in self._kept.values() for b in m.values()),+ "actors": len(self._kept),+ "root": str(self.root) if self.root else None,+ }diff --git a/src/pyrtan/transport.py b/src/pyrtan/transport.py index febef0c..1943948 100644 --- a/src/pyrtan/transport.py +++ b/src/pyrtan/transport.py@@ -129,7 +129,7 @@ class UrllibTransport:``public`` (refuse private, loopback, link-local and metadata addresses, on the first URL and on every redirect)."""- def __init__(self, user_agent: str = "pyrtan/0.4 (+https://github.com/b-i-t-8/pyrtan)", egress: str = "any") -> None:+ def __init__(self, user_agent: str = "pyrtan/0.5 (+https://github.com/b-i-t-8/pyrtan)", egress: str = "any") -> None:if egress not in EGRESS_POLICIES: raise ValueError(f"egress policy must be one of {EGRESS_POLICIES}, got {egress!r}") self.user_agent = user_agent diff --git a/tests/test_federation.py b/tests/test_federation.py index 7d47ab6..bcdfa21 100644 --- a/tests/test_federation.py +++ b/tests/test_federation.py@@ -448,3 +448,189 @@ def test_snapshot_bakes_and_bootstraps(two, tmp_path, monkeypatch, capsys):assert code == 0 and "snapshot as of" in o and b.url in o and "gone" in o monkeypatch.setenv("PYRTAN_SNAPSHOT", str(tmp_path / "nope.json")) assert cli("directory")[0] == 3+++# ---------------------------------------------------------------- members and keeping+++class Member:+ """A browser or a phone, in Python: no inbox, a mailbox it polls."""++ def __init__(self, hub_url: str, name: str = "phone"):+ import secrets as _secrets++ self.id = f"urn:pyrtan:member:{_secrets.token_hex(16)}"+ self.secret = _secrets.token_hex(32)+ self.hub = hub_url.rstrip("/")+ self.name = name+ self.kept: dict[str, dict] = {}+ self.directory: list[dict] = []++ def post(self, activity: dict, secret: str | None = None):+ activity = {"id": f"{self.id}/activities/{len(self.kept)}", "actor": self.id, "published": "now", **activity}+ return http_json("POST", f"{self.hub}/inbox", activity, headers=sign(secret or self.secret, self.id, "POST", "/inbox", activity))++ def follow(self, secret: str | None = None):+ return self.post({"type": "Follow", "object": self.hub, "secret": self.secret, "inbox": "mailbox", "name": self.name}, secret)++ def poll(self) -> list[dict]:+ status, body = http_json("GET", f"{self.hub}/mailbox", headers=sign(self.secret, self.id, "GET", "/mailbox", None))+ assert status == 200, body+ acts = body["activities"]+ for a in acts:+ if a["type"] == "Keep":+ self.kept[a["object"]["key"]] = a["object"]+ self.post({"type": "Kept", "object": {"key": a["object"]["key"], "sha256": a["object"]["sha256"]}})+ elif a["type"] == "Fetch":+ blob = self.kept.get(a["object"]["key"])+ if blob:+ self.post({"type": "Kept", "object": {"key": blob["key"], "sha256": blob["sha256"], "data": blob["data"]}})+ elif a["type"] == "Announce":+ self.directory = a["object"].get("directory", [])+ return acts++ def announce(self, listed: bool = True):+ own = [{"id": self.id, "name": self.name, "kind": "member", "trees": [], "seen": time.time()}] if listed else []+ return self.post({"type": "Announce", "object": {"type": "Capabilities", "name": self.name, "trees": [], "directory": own, "kept": sorted(self.kept)}})+++def test_member_follows_through_a_mailbox_and_keeps_blobs(two, tmp_path):+ a, b = two+ a.fed.store = __import__("pyrtan.store", fromlist=["BlobStore"]).BlobStore(tmp_path / "store-a")+ m = Member(a.url)+ status, out = m.follow()+ assert status == 202 and out["mailbox"] == f"{a.url}/mailbox"+ assert a.fed.followers[m.id].kind == "member" and a.fed.followers[m.id].inbox == "mailbox"+ # The Accept and the first Announce wait in the mailbox; polling drains it.+ acts = m.poll()+ assert [x["type"] for x in acts] == ["Accept", "Announce"] and m.poll() == []+ _, peers = a.get("/peers")+ assert peers["followers"][0]["kind"] == "member" and peers["followers"][0]["link"] == "up"+ # The member lists itself by announcing upward; the hub's directory carries it as a member.+ status, out = m.announce()+ assert status == 200 and out["directory"] is not None+ entry = next(h for h in a.fed.directory()["hubs"] if h["id"] == m.id)+ assert entry["kind"] == "member" and entry["via"] == m.id+ # A blob of the hub's own goes to the member through its mailbox; the member's Kept confirms it.+ status, out = a.post("/keep", {"key": "notes/today", "data": {"todo": ["water plants"]}})+ assert status == 200 and out["keeping"]["holders"] == [] # nobody has confirmed yet+ acts = m.poll()+ assert [x["type"] for x in acts] == ["Keep"] and m.kept["notes/today"]["data"] == {"todo": ["water plants"]}+ assert a.fed.holders("notes/today") == [m.id]+ _, out = a.get("/keep")+ assert out["keeping"][0]["key"] == "notes/today" and out["keeping"][0]["holders"] == [m.id] and out["replicas"] == 2+ # Fetch: the hub still has its own copy, so no round trip.+ assert a.post("/fetch", {"key": "notes/today"})[1]["source"] == "own"+ # The hub loses its store; the member's Announce says what it keeps, the hub asks, the member answers, the blob is back.+ a.fed.store = __import__("pyrtan.store", fromlist=["BlobStore"]).BlobStore(None)+ assert a.get("/keep/notes/today")[0] == 404+ _, out = a.post("/fetch", {"key": "notes/today"})+ assert out["status"] == "asked" and out["asked"] == [m.id]+ acts = m.poll()+ assert [x["type"] for x in acts] == ["Fetch"]+ _, out = a.get("/keep/notes/today")+ assert out["data"] == {"todo": ["water plants"]} and out["holders"] == [m.id]+ # Or without asking: the member's next upward Announce names the key, the hub notices it has no copy and fetches.+ a.fed.store = __import__("pyrtan.store", fromlist=["BlobStore"]).BlobStore(None)+ m.announce()+ assert [x["type"] for x in m.poll()] == ["Fetch"]+ assert a.get("/keep/notes/today")[1]["data"] == {"todo": ["water plants"]}+ # Leaving takes the mailbox and the kept blobs with it.+ assert m.post({"type": "Undo", "object": {"type": "Follow", "actor": m.id, "object": a.url}})[0] == 200+ assert m.id not in a.fed.followers and m.id not in a.fed.mailboxes+ assert http_json("GET", f"{a.url}/mailbox", headers=sign(m.secret, m.id, "GET", "/mailbox", None))[0] == 401+ # A member's id has a shape; a stranger's signature opens nothing.+ bad = Member(a.url); bad.id = "urn:pyrtan:member:short"+ assert bad.follow()[0] == 400+ assert http_json("GET", f"{a.url}/mailbox", headers=sign("x", "urn:pyrtan:member:" + "0" * 32, "GET", "/mailbox", None))[0] == 401+++def test_hubs_keep_for_each_other_and_the_store_has_limits(two, tmp_path):+ from pyrtan.store import MAX_BLOB, BlobStore, StoreError++ a, b = two+ a.fed.store = BlobStore(tmp_path / "a"); b.fed.store = BlobStore(tmp_path / "b")+ a.post("/follow", {"url": b.url})+ # A keeps a blob; B (a hub A follows) is a keeper and confirms in the same round trip.+ status, out = a.post("/keep", {"key": "gaps-archive", "data": [{"tree": "x"}]})+ assert status == 200 and out["keeping"]["holders"] == [b.url]+ assert b.fed.store.list_kept(a.url)[0]["key"] == "gaps-archive"+ assert [k["key"] for k in b.get("/keep")[1]["kept"]] == ["gaps-archive"]+ # A takes it back from B over the kept path, signed; a stranger cannot.+ path = "/kept/gaps-archive"+ status, blob = http_json("GET", f"{b.url}{path}", headers=sign(a.fed.following[b.url].secret, a.url, "GET", path, None))+ assert status == 200 and blob["data"] == [{"tree": "x"}]+ assert http_json("GET", f"{b.url}{path}", headers=sign("nope", a.url, "GET", path, None))[0] == 401+ assert http_json("GET", f"{b.url}/kept/other", headers=sign(a.fed.following[b.url].secret, a.url, "GET", "/kept/other", None))[0] == 404+ # Fetch after A's store is gone: B answers at once.+ a.fed.store = BlobStore(tmp_path / "a2")+ _, out = a.post("/fetch", {"key": "gaps-archive"})+ assert out["status"] == "ok" and out["source"] == b.url and out["blob"]["data"] == [{"tree": "x"}]+ # The automatic blobs appear on the poll and change when the directory does.+ a.hub.refresh_all()+ keys = {k["key"] for k in a.get("/keep")[1]["keeping"]}+ assert {"directory", "gaps", "unanswered", "gaps-archive"} <= keys+ # Limits: a key has a shape, a blob a size, an actor a budget; and the state file remembers who holds what.+ assert a.post("/keep", {"key": "Bad Key", "data": 1})[0] == 400+ assert a.post("/keep", {"key": "big", "data": "x" * (MAX_BLOB + 1)})[0] == 400+ with pytest.raises(StoreError):+ b.fed.store.put_kept(a.url, "k", {"x": 1}, sha256="wrong")+ state = tmp_path / "state.json"+ a.fed.state_path = state; a.fed._save()+ again = Federation(a.url, state=state, store=BlobStore(tmp_path / "a2"))+ assert again.keeping["gaps-archive"]["holders"][b.url]["sha256"] == a.fed.keeping["gaps-archive"]["sha256"]+ # Unfollow drops what B kept for A.+ a.post("/unfollow", {"url": b.url})+ assert b.fed.store.list_kept(a.url) == []+++def test_replication_counts_fresh_holders_only(two, tmp_path):+ from pyrtan.store import BlobStore++ a, b = two+ a.fed.store = BlobStore(None); a.fed.replicas = 2+ m1, m2, m3 = Member(a.url, "m1"), Member(a.url, "m2"), Member(a.url, "m3")+ for m in (m1, m2, m3):+ m.follow(); m.poll()+ a.post("/keep", {"key": "k", "data": 1})+ # Two members got a Keep (replicas = 2), the third did not.+ got = [m for m in (m1, m2, m3) if any(x["type"] == "Keep" for x in m.poll())]+ assert len(got) == 2 and a.fed.holders("k") == sorted(m.id for m in got)+ assert a.fed.replicate() == 0 # enough holders: nothing sent+ # A holder goes stale: the next replicate sends to the third.+ a.fed.keeping["k"]["holders"][got[0].id]["at"] -= 2 * 86400+ assert a.fed.replicate() == 1+ third = next(m for m in (m1, m2, m3) if m not in got)+ assert [x["type"] for x in third.poll()] == ["Keep"] and len(a.fed.holders("k")) == 2+ # A new version of the blob needs confirming again.+ a.post("/keep", {"key": "k", "data": 2})+ assert a.fed.holders("k") == []+ for m in (m1, m2, m3):+ m.poll()+ assert len(a.fed.holders("k")) == 2+++def test_cli_keep_kept_fetch(two, tmp_path, capsys):+ from pyrtan.store import BlobStore++ a, b = two+ a.fed.store = BlobStore(None)+ a.post("/follow", {"url": b.url})++ def run(*argv):+ code = main(list(argv))+ o = capsys.readouterr()+ return code, o.out, o.err++ f = tmp_path / "blob.json"; f.write_text(json.dumps({"hello": "world"}))+ code, out, _ = run("keep", "recipes/soup", str(f), "--at", a.url, "--token", "token-a")+ assert code == 0 and "keeps recipes/soup" in out and "held by 1 of 2" in out+ code, out, _ = run("kept", "--at", a.url, "--token", "token-a")+ assert code == 0 and "recipes/soup" in out and b.url in out+ code, out, _ = run("kept", "--at", b.url, "--token", "token-b")+ assert f"for {a.url}: recipes/soup" in out+ a.fed.store = BlobStore(None)+ code, out, _ = run("fetch", "recipes/soup", "--at", a.url, "--token", "token-a", "--out", str(tmp_path / "back.json"))+ assert code == 0 and f"from {b.url}" in out and json.loads((tmp_path / "back.json").read_text()) == {"hello": "world"}+ assert run("fetch", "nothing", "--at", a.url, "--token", "token-a")[0] == 1+ assert run("keep", "bad key", str(f), "--at", a.url, "--token", "token-a")[0] == 1diff --git a/uv.lock b/uv.lock index 724197d..ee71beb 100644 --- a/uv.lock +++ b/uv.lock@@ -147,7 +147,7 @@ wheels = [[[package]] name = "pyrtan"-version = "0.4.0"+version = "0.5.0"source = { editable = "." } [package.optional-dependencies]