Federate hubs in the shape of ActivityPub, and release 0.3.0
Claude · 2026-10-25 · 2aaf6bde0a878502db81cc2b87d4829b5d8424f7 · parents 67c0b95
A hub started with --federate (or --advertise URL) has an identity other hubs can follow. federation.py gives it an actor document at /actor, an inbox for Follow, Accept, Reject, Undo and Announce activities, and a pair secret per relationship; every hub-to-hub message carries an actor, a date and an HMAC-SHA256 signature over the method, path, date and the canonical JSON body, which is what goes on the wire.
Following is trust, in one direction. A hub routes only to the hubs its operator chose to follow (pyrtan follow, --follow, or POST /follow with the token); a follower receives announcements and may ask, and is never routed to. Federation is one hop: a hub announces what its own services hold, never what it learned from others, and refuses a request that has already crossed a hub. A followed hub fetches the follower's own /actor before accepting, so a Follow cannot point it at a stranger's inbox. --federation-secret takes follows only from hubs that sign with it; --closed takes none; --state FILE keeps the relationships across restarts. A followed hub announces whenever its capabilities change and is polled (signed) as a fallback, so its link is up, stale or down like a service's. A federated answer carries via: [the hub it crossed].
pyrtan follow, unfollow and peers drive it; pyrtan hub gains --name, --federate, --advertise, --follow, --federation-secret, --closed and --state. /peers lists following and followers, and the console's Peers tab shows them. Design in _meta/federation.md; the sequencing moves more forms of I/O to v0.4 and the catalog's growth to v0.5.
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01YChr78vDHtQJ15HNbsaRLZ
Changes
README.md | 46 ++++++- _meta/design.md | 17 ++- _meta/federation.md | 182 ++++++++++++++++++++++++++++ _meta/platform.md | 7 +- deploy/README.md | 8 ++ deploy/pyrtan.com/README.md | 8 +- deploy/pyrtan.com/catalog.yml | 2 +- deploy/pyrtan.com/site.yml | 2 +- pyproject.toml | 2 +- src/pyrtan/__init__.py | 2 +- src/pyrtan/catalog.py | 2 +- src/pyrtan/cli.py | 93 ++++++++++++++- src/pyrtan/federation.py | 402 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ src/pyrtan/hub.py | 187 +++++++++++++++++++++++------ src/pyrtan/page.py | 19 +++ src/pyrtan/service.py | 34 ++++-- src/pyrtan/transport.py | 2 +- tests/test_federation.py | 265 +++++++++++++++++++++++++++++++++++++++++ uv.lock | 2 +- 19 files changed, 1206 insertions(+), 76 deletions(-)
diff --git a/README.md b/README.md index 40ca301..0f2ef3d 100644 --- a/README.md +++ b/README.md@@ -124,7 +124,8 @@ whole format is in [`_meta/tree-format.md`](_meta/tree-format.md).| `pyrtan run --tree NAME --input k=v` | Runs a named tree. | | `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`. |+| `pyrtan hub --port 8700` | Routes between services. Services register with `--hub URL`; peers carry a `link` of `up`, `stale` or `down`. `--federate` lets other hubs follow it and it follow them. |+| `pyrtan follow URL --at HUB` | Makes your hub follow another: it then routes sentences to that hub's trees too. `unfollow` undoes it; `peers` lists services, hubs followed and followers. || `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. | | `pyrtan unanswered --at URL` | Every sentence no tree answered, with the tree that would: the trees to write next. `--scaffold DIR` writes them. |@@ -155,6 +156,38 @@ its `link`; and the gaps. The same console lives at`--cors https://pyrtan.com`. To run the swarm on a NAS or any Docker host, see [`deploy/README.md`](deploy/README.md).+## A federation++A hub can follow other hubs, the way one account follows another on the+fediverse, and answer for their trees too:++```+pyrtan hub --port 8700 --name alpha --federate # on one box+pyrtan hub --port 8700 --name beta --federate # on another+pyrtan follow https://beta.example --at https://alpha.example+pyrtan ask "top 3 stories on hacker news" --at https://alpha.example # answered by beta's service+```++Each federated hub has an actor document at `/actor` and an inbox; a+follow is a signed `Follow` activity, answered with `Accept` and an+`Announce` of what the followed hub's services hold, and a new `Announce`+whenever that changes. The shape is ActivityPub's; the wire format is+Pyrtan's own, signed with a secret the two hubs share, and nothing here+federates with Mastodon.++Three rules make it safe to run with strangers. **Following is trust, in+one direction:** a hub routes only to hubs its operator chose to follow,+and being followed grants nothing but announcements and the right to+ask. **Federation is one hop:** a hub announces what it holds itself,+never what it learned from others, and refuses to forward a request that+already crossed a hub, so there is nothing to loop. **Every message is+signed**, and a hub started with `--federation-secret` takes follows only+from hubs that know the secret; `--closed` takes none. `--state FILE`+keeps the relationships across restarts. The answer to a federated+sentence 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).+## The catalog Trees other people wrote are at <https://pyrtan.com/catalog/>, a directory@@ -281,7 +314,7 @@ slot refused. The whole set of rules is in``` src/pyrtan/ the package: tree, nlp, intent, template, conditions, extract, html, transport, runtime, trace, registry, gaps, grow,- service, hub, page (the console), catalog, cli+ service, hub, federation, page (the console), catalog, cliexamples/ trees that run against real public APIs, and offline against tests/fixtures/examples.json tests/ pytest; a swarm of two services behind a hub runs on localhost@@ -293,10 +326,11 @@ deploy/ Dockerfile at the root; a compose stack for OpenMediaVault,## Status-0.2.0. The core, the service, the hub, the command, the console and the-catalog client are built and tested; HTTP is the one form of I/O the core-speaks; a swarm can require a token, keep its gaps, and refuse private-addresses. Not yet built: further adapters (a browser for pages that need+0.3.0. The core, the service, the hub, the command, the console, the+catalog client, the language rules and federation between hubs are built+and tested; HTTP is the one form of I/O the core speaks; a swarm can+require a token, keep its gaps, refuse private addresses, and follow+other swarms. Not yet built: further adapters (a browser for pages that needa script to render; files, shells, serial lines, message queues), persistent storage for unanswered sentences, `pyrtan watch`, and hosting. See [`_meta/design.md`](_meta/design.md) for the runtime's design and diff --git a/_meta/design.md b/_meta/design.md index 1ad84e6..867d7ca 100644 --- a/_meta/design.md +++ b/_meta/design.md@@ -118,7 +118,9 @@ grow.py an unanswered sentence -> a proposed tree, a scaffold, a logservice.py one process, one registry, HTTP endpoints, hot reload, gaps, unanswered hub.py peers, link up/stale/down, routing by capability, forwarding-cli.py check · list · run · match · ask · serve · hub · gaps · unanswered · new+federation.py actor, inbox, Follow/Accept/Undo/Announce, signed; one hop+cli.py check · list · run · match · ask · serve · hub · follow · unfollow ·+ peers · gaps · unanswered · new · search · pull · index · console``` **Reading a sentence** (`nlp.py`, `intent.py`) is rules all the way down,@@ -275,17 +277,22 @@ Five trees, all against real public APIs, all tested offline:`pyrtan watch` that re-runs a set of sentences on a schedule and reports when a path changes (the way to learn a site has changed before a person does).-3. **v0.3: more forms of I/O.** First the browser: `observe` with+3. **v0.3, built: sentences read by rules, and hubs that follow each+ other.** The language rules and typed slots (§3, "Reading a+ sentence"), and federation in the shape of ActivityPub: a hub follows+ other hubs and routes to them, one hop, every message signed, following+ as the only trust. Design in [`federation.md`](federation.md).+4. **v0.4: 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.-4. **v0.4: the catalog grows up.** Namespaces, publishing through pull+5. **v0.5: 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.-5. **Beside the swarm, never inside it:** an authoring aid that reads gaps+ hubs back to a tree's author, carried as another federation topic.+6. **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 new file mode 100644 index 0000000..1278fc3 --- /dev/null +++ b/_meta/federation.md@@ -0,0 +1,182 @@+# Design — Federation: hubs that follow each other++| | |+|---|---|+| **Status** | **Built, v0.3.** `src/pyrtan/federation.py`, the hub's federation routes, `pyrtan follow` / `unfollow` / `peers`, the console's federation table, and tests of two and three hubs over real sockets. |+| **Date** | 2026-10-25 |+| **Decision** | A hub may follow other hubs, in the shape of ActivityPub (actor, inbox, Follow/Accept/Undo/Announce), and route sentences to the hubs it follows. Following is trust, in one direction, and federation is one hop. |+| **Where** | `src/pyrtan/federation.py`, `src/pyrtan/hub.py`. |++## 0. The brief++*"Make Pyrtan federated, like ActivityPub / pubsub."* Read as: a swarm+should not stop at one operator's hub. Two people who each run a hub+should be able to let their swarms answer each other's sentences, without+either handing the other a token to their whole hub, without a central+registry of hubs, and with the same trace at the end.++The ActivityPub reading: a hub is an *actor* with an *inbox*; one hub+*follows* another; the followed hub *announces* to its followers+(pubsub: capabilities are the topic, followers are the subscribers); the+follower routes to what was announced. The parts of ActivityPub this+borrows are the parts that are about who hears from whom. The parts it+does not borrow are JSON-LD, HTTP Signatures with public keys, and+interoperation with the fediverse; see §4.++## 1. What this design refuses++- **Relaying.** A hub announces what its own services hold, never what it+ learned from the hubs it follows, and a request that has already crossed+ a hub is refused rather than forwarded again. Federation is one hop+ (`MAX_HOPS = 1`). There is no loop to detect because there is no path+ long enough to loop.+- **Routing to anyone the operator did not choose.** A hub routes only to+ hubs it follows, and following is an operator's act (`pyrtan follow`,+ `--follow`, or `POST /follow` with the hub's token). Being followed+ grants nothing: a follower receives announcements and may ask, but is+ never routed to.+- **A key infrastructure.** The standard library has HMAC and nothing+ asymmetric. A pair secret, made by the follower and carried in the+ Follow, is enough for two hubs whose operators chose each other. A+ shared federation secret turns an open hub into a closed group.+- **Trusting an inbox a stranger named.** A Follow says where to send+ Accepts and Announces; the followed hub fetches the follower's own+ `/actor` and takes the Follow only if that document names the same+ inbox. A Follow cannot point a hub at someone else's server.++## 2. The protocol++Every hub started with `--federate` (or `--advertise URL`) has an+identity, `id`, which is the URL other hubs reach it at.++| | Method, path | Who | What |+| :-- | :-- | :-- | :-- |+| Actor | `GET /actor` | anyone | `{type: "Hub", id, name, inbox, capabilities, follows: "open" \| "secret"}` |+| Inbox | `POST /inbox` | signed | an activity: `Follow`, `Accept`, `Reject`, `Undo`, `Announce` |+| Follow | `POST /follow {url}` | operator | start following `url` |+| Unfollow | `POST /unfollow {url}` | operator | stop |+| Who | `GET /following`, `/followers`, `/peers` | operator | the relationships, each with `state` and `link` |+| Log | `GET /outbox` | operator | the last fifty activities sent and received, with delivery results |++**Follow.** A (the follower) fetches `B/actor`, makes a 32-byte pair+secret, records B as `pending`, and posts to B's inbox:++```json+{"type": "Follow", "id": "A/activities/…", "actor": "A", "object": "B",+ "secret": "<pair secret>", "inbox": "A/inbox", "name": "alpha-hub"}+```++signed with B's federation secret if A was given one, else with the pair+secret (which proves nothing yet, but keeps every message signed).++**Accept.** B checks its policy: with a federation secret, the Follow+must be signed with it (401 otherwise); without one, open follows are+taken unless `--closed` (403). B fetches `A/actor` and refuses unless+`id` and `inbox` match (400). B records A as a follower with the pair+secret, then posts to A's inbox an `Accept` and an `Announce`, both+signed with the pair secret. A marks B `accepted`. The whole handshake is+synchronous inside A's `POST /follow`, so `pyrtan follow` reports the+outcome.++**Announce** (the pubsub). `{"type": "Announce", "object": {"type":+"Capabilities", "name", "trees": [...]}}`, the same shape as+`/capabilities`, sent to every follower whenever what the hub's own+services hold changes (checked on every poll by a hash of the trees'+names, versions, triggers, inputs, language and hosts). A follower also+polls `B/capabilities` (signed) on its own poll interval, so B's `link`+is `up`, `stale` or `down` by the same clock as a service's.++**Undo.** `{"type": "Undo", "object": {"type": "Follow", …}}` from the+follower ends the relationship on both sides.++**Signatures.** Three headers on every hub-to-hub request:+`X-Pyrtan-Actor` (the sender's id), `X-Pyrtan-Date` (unix seconds, within+five minutes), `X-Pyrtan-Signature` (HMAC-SHA256 over+`METHOD\npath\ndate\nsha256(canonical body)`). The body on the wire *is*+the canonical JSON (sorted keys, no spaces), so the receiver verifies from+what it parsed. A signed request may reach only `/inbox`, `/run`,+`/intent`, `/resume` and `/capabilities`; the operator's paths still need+the bearer token.++**Which secret.** With a mutual follow there are two pairs. A request is+signed with the secret of the relationship in which the requester follows+the target, because asking is what following permits; the target+verifies with its followers' secrets first.++## 3. Routing++`Hub.holders()` is the hub's services followed by the hubs it follows;+`candidates()` reads a sentence against every holder's triggers with the+same rules as before (`intent.py`, with each tree's `[language]`). On a+tie between two holders of the *same* tree name, `_rank` prefers a+service of ours over a hub we follow, then the healthier `link`. A tie+between different tree names is `ambiguous`, as it always was.++Forwarding to a hub we follow goes to its `/run`, signed, with+`hops: [our id]`. The answer comes back with `peer` (the service that+ran it, as the remote hub set it) and `via: [the remote hub]`. The+reading step at the top of the trace is ours; the rest is the remote's.++A hub receiving `/run` with `hops` already set routes only to its own+services; if the tree is held by a hub *it* follows, it answers+`hop-limit` and says so. `pyrtan run --tree NAME` on a hub that only+knows the tree through two hops answers `no-such-tree`, because the hub+never learned of it: nothing is announced that was not held.++## 4. Paths considered and not taken++**ActivityPub proper.** JSON-LD, `@context`, HTTP Signatures with RSA+keys, WebFinger, the Activity Streams vocabulary. Each would be a+dependency (there is no JSON-LD or RSA in the standard library) and none+would buy interoperation that means anything: a Mastodon server cannot+run a tree. The shape is kept because it is the right shape for "who+hears from whom"; the wire format is Pyrtan's own and says so.++**A directory of hubs.** A place where hubs register so others can find+them. Rejected for now: it is the catalog's job to make trees findable,+and a hub's job to answer for what it holds. An operator who wants to be+followed publishes their hub's URL; the catalog could carry "hubs that+hold this tree" later, as metadata, without the runtime changing.++**Multi-hop routing.** Announcing transitive capabilities with a hop+count, and forwarding with a TTL. It would let a sentence cross three+hubs, and it would make "why did this go there" a path through other+people's follow lists. One hop keeps every route explainable by one+operator's own choices. `MAX_HOPS` is a constant so the decision is+visible, not so it is cheap to change.++**Sharing gaps and unanswered sentences over federation.** The natural+next topic for the pubsub, and the one that makes the catalog loop close+(platform.md §3, phase 2). Not in v0.3: the shape is settled (another+`Announce` object type), the question is what a hub does with a+stranger's gaps.++## 5. Running it++```+# hub A, reachable at https://a.example (through your TLS proxy)+pyrtan hub --host 0.0.0.0 --token "$A_TOKEN" --advertise https://a.example --state /var/lib/pyrtan/federation.json++# hub B likewise, then on A's operator's shell:+pyrtan follow https://b.example --at https://a.example --token "$A_TOKEN"+pyrtan peers --at https://a.example --token "$A_TOKEN"+pyrtan ask "a sentence only B's trees answer" --at https://a.example --token "$A_TOKEN"+```++To let only hubs you know follow yours, start it with+`--federation-secret` (or `$PYRTAN_FEDERATION_SECRET`) and give the+secret to those operators; their hubs must carry the same secret to+follow. `--closed` takes no follows at all while still following others.+`--state FILE` keeps the relationships across restarts; without it a+restart forgets them and the other side's `link` goes `down` until the+follow is redone.++## 6. What this design does not claim++It 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+secret is private from its followers, only from strangers. It does not+claim to defend against a followed hub that lies in its announcements+(it can claim to answer "weather" and will be asked); following is+trust, and the trace shows `via` so the lie has a name. It does not+claim to be ActivityPub.diff --git a/_meta/platform.md b/_meta/platform.md index 3db1807..62c2c4f 100644 --- a/_meta/platform.md +++ b/_meta/platform.md@@ -121,7 +121,12 @@ says so and points at the hub's own `/`, which is the same console.A tenant gets a hub and one or more services as containers from this repository's image, with a token, `--cors https://pyrtan.com`, `--gaps` on-a volume, `--egress public`, and TLS from the ingress. The control plane+a volume, `--egress public`, `--federate --advertise https://<tenant>.pyrtan.com`+with `--state` on the volume, and TLS from the ingress. A hosted hub and+a self-hosted one follow each other like any two hubs+([`federation.md`](federation.md)): a customer's NAS can answer for their+hosted swarm and the other way round, which is the invariant in §0 made+concrete. The control plane(separate, private) does what the runtime does not: sign-up and teams, provisioning, the secrets that reach trees as `env`, trace retention, a scheduler that re-runs sentences and reports when a path changes (the diff --git a/deploy/README.md b/deploy/README.md index a0edc85..eb8ddf3 100644 --- a/deploy/README.md +++ b/deploy/README.md@@ -62,6 +62,14 @@ restart; the container runs as uid 10001, so `chown 10001 trees` (or`chmod o+w trees`) on the NAS, else the gaps stay in memory and `/health` reports `gaps_file_error`.+To let this swarm answer for another (a friend's hub, or a hosted one),+add `--federate --advertise https://<your-public-url> --state+/trees/.federation.json` to the hub's command once it is behind TLS, then+`pyrtan follow https://other-hub --at https://<your-public-url> --token …`.+Following is one-directional trust: your hub routes to the hubs you+follow, and hubs that follow yours may ask it. See+[`../_meta/federation.md`](../_meta/federation.md).+### Without the plugin Over SSH, from the checkout: diff --git a/deploy/pyrtan.com/README.md b/deploy/pyrtan.com/README.md index 7cc0fed..0b5d291 100644 --- a/deploy/pyrtan.com/README.md +++ b/deploy/pyrtan.com/README.md@@ -143,9 +143,11 @@ which is the same console served locally.## 3. Later: hosting Nothing here needs to change. A tenant's hub is this repository's image-started with `--token`, `--cors https://pyrtan.com`, `--gaps` on a volume-and `--egress public`, behind TLS; the console at pyrtan.com connects to it-like to any other. The control plane that provisions it is a separate,+started with `--token`, `--cors https://pyrtan.com`, `--gaps` on a volume,+`--egress public` and `--federate --advertise https://<tenant>.pyrtan.com+--state <volume>/federation.json`, behind TLS; the console at pyrtan.com+connects to it like to any other, and a self-hosted hub can follow it (or+be followed by it) like any other hub. The control plane that provisions it is a separate,private repository, per `_meta/platform.md` §2.6. ## Checklist diff --git a/deploy/pyrtan.com/catalog.yml b/deploy/pyrtan.com/catalog.yml index 91bf3d9..4d0dba4 100644 --- a/deploy/pyrtan.com/catalog.yml +++ b/deploy/pyrtan.com/catalog.yml@@ -15,7 +15,7 @@ on:pull_request: env:- PYRTAN_REF: v0.2.0+ PYRTAN_REF: v0.3.0permissions: contents: read diff --git a/deploy/pyrtan.com/site.yml b/deploy/pyrtan.com/site.yml index 451bbd4..2a19cc3 100644 --- a/deploy/pyrtan.com/site.yml +++ b/deploy/pyrtan.com/site.yml@@ -12,7 +12,7 @@ on:workflow_dispatch: env:- PYRTAN_REF: v0.2.0+ PYRTAN_REF: v0.3.0permissions: contents: read diff --git a/pyproject.toml b/pyproject.toml index 115dfed..5d2224a 100644 --- a/pyproject.toml +++ b/pyproject.toml@@ -1,6 +1,6 @@[project] name = "pyrtan"-version = "0.2.0"+version = "0.3.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/src/pyrtan/__init__.py b/src/pyrtan/__init__.py index ee50ff2..3a0c2e0 100644 --- a/src/pyrtan/__init__.py +++ b/src/pyrtan/__init__.py@@ -33,4 +33,4 @@ __all__ = ["Response", ]-__version__ = "0.2.0"+__version__ = "0.3.0"diff --git a/src/pyrtan/catalog.py b/src/pyrtan/catalog.py index 2cd8bcb..b474fd0 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.2", "Accept": "application/json, application/toml, text/plain, */*"})+ req = urllib.request.Request(url, headers={"User-Agent": "pyrtan-catalog/0.3", "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 43cb98d..84bfe69 100644 --- a/src/pyrtan/cli.py +++ b/src/pyrtan/cli.py@@ -8,6 +8,10 @@pyrtan ask TEXT --at URL answer a sentence through a service or hub pyrtan serve [PATH ...] [--port N] host trees as a service pyrtan hub [--peer URL ...] route between services+ pyrtan hub --federate --advertise URL [--follow URL ...] and between hubs+ pyrtan follow URL --at HUB follow another hub (and route to it)+ pyrtan unfollow URL --at HUB+ pyrtan peers --at HUB services, and hubs followed and followingpyrtan 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@@ -99,11 +103,29 @@ def main(argv: list[str] | None = None) -> int:p = sub.add_parser("hub", help="route between services") p.add_argument("--host", default="127.0.0.1") p.add_argument("--port", type=int, default=8700)+ p.add_argument("--name", default="hub", help="what this hub calls itself, to the console and to hubs that follow it")p.add_argument("--peer", action="append", default=[], help="a service URL to start with (repeatable; services may also register themselves)") p.add_argument("--poll", type=float, default=10.0, help="seconds between health polls") _access_args(p)+ p.add_argument("--federate", action="store_true", help="answer /actor and /inbox so other hubs can follow this one, and follow others")+ p.add_argument("--advertise", help="the URL other hubs reach this hub at (implies --federate; default: this host and port)")+ p.add_argument("--follow", action="append", default=[], metavar="URL", help="a hub to follow on start (repeatable)")+ p.add_argument("--federation-secret", default=os.environ.get("PYRTAN_FEDERATION_SECRET"), help="only hubs that sign their Follow with this secret may follow (default $PYRTAN_FEDERATION_SECRET; without it, follows are open)")+ 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", help="keep who this hub follows and is followed by in this file, across restarts")p.add_argument("--quiet", action="store_true")+ for verb in ("follow", "unfollow"):+ p = sub.add_parser(verb, help=f"{verb} another hub, through a running federated hub of yours")+ p.add_argument("url", help="the other hub's URL")+ 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("peers", help="a hub's services, and the hubs it follows and is followed by")+ 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, "the hub's token, if any")+p = sub.add_parser("gaps", help="decisions that met a value with no branch") p.add_argument("--at", help="a running service or hub") p.add_argument("--json", action="store_true")@@ -143,6 +165,7 @@ def main(argv: list[str] | None = None) -> int:commands = { "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,} return commands[args.command](args)@@ -366,10 +389,15 @@ def cmd_serve(args: argparse.Namespace) -> int:def cmd_hub(args: argparse.Namespace) -> int: from .hub import serve_hub- hub, server = serve_hub(args.host, args.port, peers=args.peer, poll_every=args.poll, quiet=args.quiet, token=args.token, cors=args.cors)- print(f"pyrtan hub at {server.url}: {len(hub.peers)} peer(s)" + _access_note(args), flush=True)+ hub, server = serve_hub(+ args.host, args.port, name=args.name, peers=args.peer, poll_every=args.poll, quiet=args.quiet, token=args.token, cors=args.cors,+ federate=args.federate or bool(args.advertise) or bool(args.follow), advertise=args.advertise, federation_secret=args.federation_secret,+ state=args.state, open_follows=not args.closed, follow=args.follow,+ )+ fed = f", federated as {hub.federation.id}" + (f" following {len(hub.federation.following)}" if hub.federation.following else "") if hub.federation else ""+ print(f"pyrtan hub at {server.url}: {len(hub.peers)} peer(s){fed}" + _access_note(args), flush=True)try:- server.serve_forever()+ server.wait()except KeyboardInterrupt: pass finally:@@ -404,6 +432,61 @@ def cmd_gaps(args: argparse.Namespace) -> int:return 0+def cmd_follow(args: argparse.Namespace) -> int:+ return _hub_post(args, "/follow", {"url": args.url}, lambda b: f"{args.at} now follows {b['following']['url']} ({b['following']['state']}); it routes there once accepted")+++def cmd_unfollow(args: argparse.Namespace) -> int:+ return _hub_post(args, "/unfollow", {"url": args.url}, lambda b: f"{args.at} no longer follows {args.url}" if b.get("removed") else f"{args.at} was not following {args.url}")+++def _hub_post(args: argparse.Namespace, path: str, payload: dict[str, Any], said) -> int:+ at = args.at.rstrip("/")+ try:+ status, body = http_json("POST", f"{at}{path}", payload, 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 == 404:+ print(f"pyrtan: {at} is not a federated hub; start it with --federate", file=sys.stderr)+ return 3+ if status != 200 or not isinstance(body, dict) or body.get("status") != "ok":+ print(f"pyrtan: {at} answered {status}: {(body or {}).get('say') if isinstance(body, dict) else body}", file=sys.stderr)+ return 1+ print(said(body))+ return 0+++def cmd_peers(args: argparse.Namespace) -> int:+ at = args.at.rstrip("/")+ try:+ status, body = http_json("GET", f"{at}/peers", 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 args.json:+ print(json.dumps(body, indent=2, ensure_ascii=False))+ return 0+ for p in body.get("peers", []):+ print(f"{p['link']:8} service {p['name'] or '-':12} {p['url']} {', '.join(p.get('trees', []))}")+ for r in body.get("following", []):+ print(f"{r['link']:8} follows {r['name'] or '-':12} {r['url']} {', '.join(r.get('trees', []))}" + (f" ({r['last_error']})" if r.get("last_error") else ""))+ for r in body.get("followers", []):+ print(f"{r['link']:8} follower {r['name'] or '-':12} {r['url']}")+ if not any(body.get(k) for k in ("peers", "following", "followers")):+ print("no peers")+ return 0++def _access_note(args: argparse.Namespace) -> str: notes = [] if getattr(args, "token", None):@@ -412,6 +495,10 @@ def _access_note(args: argparse.Namespace) -> str:notes.append("cors: " + ", ".join(args.cors)) if getattr(args, "egress", "any") != "any": notes.append(f"egress: {args.egress}")+ if getattr(args, "federation_secret", None):+ notes.append("follows need the federation secret")+ elif getattr(args, "closed", False):+ notes.append("no follows taken")return f" ({'; '.join(notes)})" if notes else "" diff --git a/src/pyrtan/federation.py b/src/pyrtan/federation.py new file mode 100644 index 0000000..6d3fd23 --- /dev/null +++ b/src/pyrtan/federation.py@@ -0,0 +1,402 @@+"""Federation: hubs that follow each other.++A hub on its own routes sentences to the services that registered with+it. A federated hub also *follows* other hubs, the way one account follows+another on the fediverse, and answers for their trees too. The shape is+ActivityPub's (an actor document, an inbox, ``Follow`` / ``Accept`` /+``Undo`` / ``Announce`` activities) without being an implementation of it:+the activities are Pyrtan's own, the signatures are HMAC on a secret the+two hubs share, and nothing here federates with Mastodon.++The one rule everything rests on: **following is trust, in one direction.**++- A hub routes sentences only to hubs it follows. Its operator chose them.+- A hub that follows another may ask it to run trees; the followed hub+ serves its followers. Nobody else can ask over federation.+- Being followed grants nothing: a follower receives announcements and may+ ask, but the followed hub never routes anything to it.+- A hub announces what its own services hold, never what it learned from+ the hubs it follows, and does not relay: federation is one hop. A+ request that has already crossed a hub is refused rather than forwarded+ again, so there are no loops to detect.++How it works. Each hub has an *actor* document at ``GET /actor`` (its id,+its inbox, its name). To follow, hub A fetches B's actor, makes a secret+for the pair, records B as pending, and posts a ``Follow`` carrying the+secret to B's inbox. B checks the policy (open, or a shared federation+secret the Follow must be signed with), fetches A's actor to confirm the+inbox is A's, records A as a follower, and posts back an ``Accept`` and an+``Announce`` of its capabilities, both signed with the pair secret. From+then on B announces whenever its capabilities change and A polls B's+``/capabilities`` (signed) as a fallback, so B's ``link`` is up, stale or+down like any peer's. ``Undo`` of the Follow ends it.++Every message between two hubs is signed: ``X-Pyrtan-Actor`` names the+sender, ``X-Pyrtan-Date`` carries a unix timestamp within five minutes,+and ``X-Pyrtan-Signature`` is HMAC-SHA256 over the method, path, date and+the SHA-256 of the canonical JSON body. The body sent *is* that canonical+JSON, so the receiver can verify from the body it parsed.++What this module refuses: open relaying (no hop beyond one), asymmetric+secrets (there is no key infrastructure to lean on in the standard+library, and a pair secret is enough for hubs whose operators chose each+other), and any routing decision that was not an operator's follow.+"""++from __future__ import annotations++import hashlib+import hmac+import json+import secrets as _secrets+import threading+import time+import uuid+from collections import deque+from dataclasses import dataclass, field+from pathlib import Path+from typing import Any, Callable++from . import __version__+from .service import bearer_ok, http_json++SIGNATURE_WINDOW = 300.0 # seconds a signed message stays valid+MAX_HOPS = 1 # federation is one hop; raise with care+STALE_AFTER = 30.0+DOWN_AFTER = 90.0+ACTIVITY_TYPES = ("Follow", "Accept", "Reject", "Undo", "Announce")+SIGNED_PATHS = ("/inbox", "/run", "/intent", "/resume", "/capabilities")+++# ---------------------------------------------------------------- signing+++def canonical(body: Any) -> bytes:+ return json.dumps(body, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode("utf-8")+++def sign(secret: str, actor: str, method: str, path: str, body: Any, date: float | None = None) -> dict[str, str]:+ date_text = str(int(date if date is not None else time.time()))+ digest = hashlib.sha256(canonical(body) if body is not None else b"").hexdigest()+ message = "\n".join([method.upper(), path, date_text, digest]).encode("utf-8")+ mac = hmac.new(secret.encode("utf-8"), message, hashlib.sha256).hexdigest()+ return {"X-Pyrtan-Actor": actor, "X-Pyrtan-Date": date_text, "X-Pyrtan-Signature": mac}+++def verify(secret: str, method: str, path: str, body: Any, headers: dict[str, str], now: float | None = None) -> str | None:+ """None if the signature holds; otherwise what is wrong with it."""+ date_text = headers.get("X-Pyrtan-Date", "")+ mac = headers.get("X-Pyrtan-Signature", "")+ if not date_text.isdigit() or not mac:+ return "no signature"+ if abs((now if now is not None else time.time()) - int(date_text)) > SIGNATURE_WINDOW:+ return "signature too old"+ expected = sign(secret, headers.get("X-Pyrtan-Actor", ""), method, path, body, int(date_text))["X-Pyrtan-Signature"]+ return None if hmac.compare_digest(expected, mac) else "bad signature"+++def _lower(headers: dict[str, str] | None) -> dict[str, str]:+ """Header lookup is case-insensitive; keep the X-Pyrtan-* names as written."""+ out: dict[str, str] = {}+ for k, v in (headers or {}).items():+ key = "-".join(p.capitalize() if p.lower() != "pyrtan" else "Pyrtan" for p in k.split("-"))+ out[key] = v+ return out+++# ---------------------------------------------------------------- remotes+++@dataclass+class Remote:+ """A hub we follow, or one that follows us."""++ url: str+ inbox: str+ secret: str+ name: str = ""+ state: str = "pending" # pending | accepted | rejected+ trees: list[dict[str, Any]] = field(default_factory=list)+ last_ok: float = 0.0+ last_error: str | None = None+ since: float = field(default_factory=time.time)+ kind: str = "hub"++ @property+ def link(self) -> str:+ if self.state != "accepted":+ return "down" if self.state == "rejected" else "pending"+ if self.last_ok == 0.0:+ return "down" if self.last_error else "pending"+ age = time.time() - self.last_ok+ if age < STALE_AFTER:+ return "up"+ if age < DOWN_AFTER:+ return "stale"+ 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}++ 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}++ @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())))+++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) -> None:+ self.id = hub_id.rstrip("/")+ self.name = name+ self.secret = secret or None # a shared federation secret: Follows must be signed with it+ self.open_follows = open_follows # without a secret, whether strangers may follow+ self.state_path = Path(state) if state else None+ self.capabilities = capabilities or (lambda: [])+ self.following: dict[str, Remote] = {}+ self.followers: dict[str, Remote] = {}+ self.outbox: deque[dict[str, Any]] = deque(maxlen=50)+ self.inbox_log: deque[dict[str, Any]] = deque(maxlen=50)+ self._announced: str | None = None+ self._lock = threading.RLock()+ if self.state_path and self.state_path.exists():+ self._load()++ # ------------------------------------------------------------ actor++ @property+ def inbox(self) -> str:+ 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", "pyrtan": __version__, "follows": "open" if (self.open_follows and not self.secret) else "secret"}++ # ------------------------------------------------------------ state++ def _save(self) -> None:+ if not self.state_path:+ return+ data = {"following": [r.to_state() for r in self.following.values()], "followers": [r.to_state() for r in self.followers.values()]}+ self.state_path.parent.mkdir(parents=True, exist_ok=True)+ self.state_path.write_text(json.dumps(data, indent=2) + "\n", encoding="utf-8")++ def _load(self) -> None:+ assert self.state_path+ try:+ data = json.loads(self.state_path.read_text(encoding="utf-8"))+ except (OSError, json.JSONDecodeError):+ return+ for d in data.get("following", []):+ self.following[d["url"]] = Remote.from_state(d)+ for d in data.get("followers", []):+ self.followers[d["url"]] = Remote.from_state(d)++ # ------------------------------------------------------------ outgoing++ 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 deliver(self, remote: Remote, activity: dict[str, Any], secret: str | None = None) -> tuple[bool, str | None]:+ """Post an activity to a remote's inbox, signed with the pair secret+ (or the one given). Returns (delivered, error)."""+ path = "/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)+ ok = status in (200, 202)+ self.outbox.append({**activity, "to": remote.url, "delivered": ok, "error": None if ok else f"{status}: {answer}"})+ return ok, None if ok else f"inbox answered {status}: {answer}"++ def follow(self, url: str) -> Remote:+ """Follow the hub at ``url``: fetch its actor, make a pair secret,+ send the Follow. The remote is pending until it Accepts."""+ url = url.rstrip("/")+ if url == self.id:+ raise ValueError("a hub cannot follow itself")+ try:+ status, actor = http_json("GET", f"{url}/actor", timeout=10.0)+ except Exception as e:+ raise ValueError(f"{url} does not answer /actor: {e}") from e+ if status != 200 or not isinstance(actor, dict) or actor.get("type") != "Hub" or not isinstance(actor.get("inbox"), str):+ raise ValueError(f"{url} does not answer /actor as a Pyrtan hub ({status})")+ with self._lock:+ remote = self.following.get(url)+ if remote is None:+ remote = Remote(url, actor["inbox"], _secrets.token_hex(32), actor.get("name", ""))+ self.following[url] = remote+ else:+ remote.inbox = actor["inbox"]+ remote.name = actor.get("name", remote.name)+ remote.state = "pending"+ self._save()+ activity = self._activity("Follow", url, secret=remote.secret, inbox=self.inbox, name=self.name)+ # 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)+ if not ok:+ remote.last_error = error+ return remote++ def unfollow(self, url: str) -> bool:+ url = url.rstrip("/")+ with self._lock:+ remote = self.following.pop(url, None)+ self._save()+ if remote is None:+ return False+ self.deliver(remote, self._activity("Undo", {"type": "Follow", "actor": self.id, "object": url}))+ return True++ def announce(self, force: bool = False) -> int:+ """Tell every follower what this hub's own services hold, when it+ changed (or ``force``). Returns how many deliveries succeeded."""+ caps = self.capabilities()+ digest = hashlib.sha256(canonical([{k: t.get(k) for k in ("name", "version", "triggers", "inputs", "language", "hosts")} for t in caps])).hexdigest()+ if digest == self._announced and not force:+ return 0+ self._announced = digest+ activity = self._activity("Announce", {"type": "Capabilities", "name": self.name, "trees": caps})+ delivered = 0+ for remote in list(self.followers.values()):+ ok, error = self.deliver(remote, activity)+ if ok:+ delivered += 1+ remote.last_ok = time.time()+ remote.last_error = None+ else:+ remote.last_error = error+ return delivered++ def refresh(self, remote: Remote) -> None:+ """Poll a hub we follow, signed: the fallback when its announces+ do not arrive, and what keeps its link honest."""+ if remote.state != "accepted":+ return+ path = "/capabilities"+ try:+ status, caps = http_json("GET", f"{remote.url}{path}", timeout=5.0, headers=sign(remote.secret, self.id, "GET", path, None))+ except Exception as e:+ remote.last_error = str(e)+ return+ if status == 200 and isinstance(caps, dict):+ self._take_capabilities(remote, caps)+ else:+ remote.last_error = f"capabilities answered {status}"++ def refresh_all(self) -> None:+ for remote in list(self.following.values()):+ self.refresh(remote)++ def _take_capabilities(self, remote: Remote, caps: dict[str, Any]) -> None:+ with self._lock:+ # Only what the hub holds itself: a tree it learned from a hub it+ # follows is two hops away and not ours to route to.+ remote.trees = [t for t in caps.get("trees", []) if isinstance(t, dict) and not t.get("via")]+ remote.name = caps.get("name") or remote.name+ remote.last_ok = time.time()+ remote.last_error = None++ # ------------------------------------------------------------ incoming++ def secret_for(self, actor: str) -> str | None:+ """The secret a request from ``actor`` is signed with. A hub asks the+ hubs it follows, so the pair in which the actor follows us comes+ first; with a mutual follow there are two pairs and this is the one+ the actor signs requests to us with."""+ actor = actor.rstrip("/")+ remote = self.followers.get(actor) or self.following.get(actor)+ return remote.secret if remote else None++ def verify_request(self, method: str, path: str, body: Any, headers: dict[str, str] | None) -> str | None:+ """For a signed request to a routed path: the actor it is from, or+ None (and the caller answers 401)."""+ h = _lower(headers)+ actor = h.get("X-Pyrtan-Actor", "").rstrip("/")+ secret = self.secret_for(actor)+ if not actor or secret is None:+ return None+ return actor if verify(secret, method, path, body, h) is None else None++ def handle_inbox(self, activity: Any, headers: dict[str, str] | None) -> tuple[int, dict[str, Any]]:+ h = _lower(headers)+ if not isinstance(activity, dict) or activity.get("type") not in ACTIVITY_TYPES or not isinstance(activity.get("actor"), str):+ return 400, {"status": "error", "reason": "bad-activity", "say": f"Send an activity of type {', '.join(ACTIVITY_TYPES)} with an actor."}+ actor = activity["actor"].rstrip("/")+ if h.get("X-Pyrtan-Actor", "").rstrip("/") != actor:+ return 401, {"status": "error", "reason": "unsigned", "say": "The signature's actor must be the activity's actor."}+ kind = activity["type"]+ self.inbox_log.append({**activity, "received": time.time()})+ if kind == "Follow":+ return self._on_follow(actor, activity, h)+ remote = self.following.get(actor) if kind in ("Accept", "Reject", "Announce") else self.followers.get(actor)+ if 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:+ return 401, {"status": "error", "reason": "unsigned", "say": problem}+ if kind == "Accept":+ with self._lock:+ remote.state = "accepted"+ remote.last_ok = time.time()+ remote.last_error = None+ self._save()+ return 200, {"status": "ok"}+ if kind == "Reject":+ with self._lock:+ remote.state = "rejected"+ remote.last_error = str(activity.get("summary") or "the hub refused the follow")+ self._save()+ return 200, {"status": "ok"}+ if kind == "Announce":+ obj = activity.get("object")+ if not isinstance(obj, dict) or not isinstance(obj.get("trees"), list):+ return 400, {"status": "error", "reason": "bad-activity", "say": "An Announce carries {\"trees\": [...]}."}+ self._take_capabilities(remote, obj)+ return 200, {"status": "ok", "trees": len(remote.trees)}+ if kind == "Undo":+ with self._lock:+ self.followers.pop(actor, None)+ self._save()+ return 200, {"status": "ok"}+ return 400, {"status": "error", "reason": "bad-activity"}++ def _on_follow(self, actor: str, activity: dict[str, Any], h: dict[str, str]) -> tuple[int, dict[str, Any]]:+ pair = activity.get("secret")+ inbox = activity.get("inbox")+ if not isinstance(pair, str) or len(pair) < 16 or not isinstance(inbox, str):+ return 400, {"status": "error", "reason": "bad-activity", "say": "A Follow carries a pair secret and the follower's inbox."}+ if self.secret:+ problem = verify(self.secret, "POST", "/inbox", activity, h)+ if problem:+ return 401, {"status": "error", "reason": "unsigned", "say": f"This hub takes follows signed with its federation secret: {problem}."}+ elif not self.open_follows:+ return 403, {"status": "error", "reason": "closed", "say": "This hub does not take follows."}+ else:+ 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."}+ with self._lock:+ remote = Remote(actor, inbox, pair, str(activity.get("name") or doc.get("name") or ""), state="accepted", last_ok=time.time())+ 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", {"type": "Capabilities", "name": self.name, "trees": self.capabilities()}))+ return 202, {"status": "ok", "say": f"{self.id} accepts {actor}."}++ # ------------------------------------------------------------ views++ def to_dict(self) -> dict[str, Any]:+ with self._lock:+ return {"id": self.id, "following": [r.to_dict() for r in self.following.values()], "followers": [r.to_dict() for r in self.followers.values()]}diff --git a/src/pyrtan/hub.py b/src/pyrtan/hub.py index 1e50d08..61d1c96 100644 --- a/src/pyrtan/hub.py +++ b/src/pyrtan/hub.py@@ -19,6 +19,11 @@ Endpoints:``GET /unanswered`` sentences no peer answered, each with a proposed tree ``GET /`` the console; ``GET /api`` is the JSON index+Federated (``--federate``), a hub also answers ``GET /actor``, ``POST+/inbox``, ``POST /follow`` and ``/unfollow`` (operator), ``GET /following``,+``/followers`` and ``/outbox``, and routes to the hubs it follows; see+:mod:`pyrtan.federation`.+A service that holds several trees and a hub that routes between services answer the same requests, so a client need not know which it is talking to, and a swarm can start as one process and split later without the callers@@ -28,18 +33,18 @@ included, so a stranger cannot join the swarm) and presents it to peers.from __future__ import annotations+import datetime as _dtimport threading import time from dataclasses import dataclass, field from typing import Any from . import __version__+from .federation import MAX_HOPS, Federation, Remote, signfrom .grow import Unanswered, no_match_outcome-import datetime as _dt-from .intent import Trigger, read from .page import PAGE-from .service import Server, http_json+from .service import Server, bearer_ok, http_jsonSTALE_AFTER = 30.0 DOWN_AFTER = 90.0@@ -53,6 +58,7 @@ class Peer:last_ok: float = 0.0 last_error: str | None = None registered: float = field(default_factory=time.time)+ kind: str = "service"@property def link(self) -> str:@@ -75,12 +81,30 @@ class Hub:self.poll_every = poll_every self.token = token or None self.peers: dict[str, Peer] = {}+ self.federation: Federation | None = Noneself.unanswered = Unanswered() self.started = time.time() self._lock = threading.RLock() self._stop = threading.Event() self._thread: threading.Thread | None = None+ # ------------------------------------------------------------ federation++ def federate(self, advertise: str, *, secret: str | None = None, state: str | None = None, open_follows: bool = True) -> 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)+ return self.federation++ def local_capabilities(self) -> list[dict[str, Any]]:+ """What this hub's own services hold: what it announces."""+ trees: list[dict[str, Any]] = []+ with self._lock:+ for peer in self.peers.values():+ for t in peer.trees:+ trees.append({**t, "peer": peer.url, "link": peer.link})+ return trees+# ------------------------------------------------------------ peers def add_peer(self, url: str, name: str = "") -> Peer:@@ -113,6 +137,9 @@ class Hub:def refresh_all(self) -> None: for peer in list(self.peers.values()): self.refresh(peer)+ if self.federation is not None:+ self.federation.refresh_all()+ self.federation.announce() # only when what we hold changeddef start_polling(self) -> None: def loop() -> None:@@ -127,28 +154,36 @@ class Hub:# ------------------------------------------------------------ lookup+ def holders(self) -> list[Peer | Remote]:+ """Everyone this hub may route to: its services, then the hubs it follows."""+ with self._lock:+ out: list[Peer | Remote] = list(self.peers.values())+ if self.federation is not None:+ out.extend(self.federation.following.values())+ return out+def candidates(self) -> list[tuple[tuple[str, str], Trigger]]: out: list[tuple[tuple[str, str], Trigger]] = []- with self._lock:- for peer in self.peers.values():- if peer.link == "down":- continue- for tree in peer.trees:- for pattern in tree.get("triggers", []):- try:- out.append(((peer.url, tree["name"]), Trigger(pattern, language=tree.get("language") or {})))- except Exception:- continue+ for holder in self.holders():+ if holder.link == "down":+ continue+ for tree in holder.trees:+ for pattern in tree.get("triggers", []):+ try:+ out.append(((holder.url, tree["name"]), Trigger(pattern, language=tree.get("language") or {})))+ except Exception:+ continuereturn out- def holder(self, tree_name: str) -> Peer | None:- with self._lock:- best: Peer | None = None- for peer in self.peers.values():- if any(t.get("name") == tree_name for t in peer.trees):- if best is None or _rank(peer) < _rank(best):- best = peer- return best+ def holder(self, tree_name: str) -> Peer | Remote | None:+ """Who to send a tree to: a service of ours before a hub we follow,+ and the healthiest of either."""+ best: Peer | Remote | None = None+ for h in self.holders():+ if any(t.get("name") == tree_name for t in h.trees):+ if best is None or _rank(h) < _rank(best):+ best = h+ return best# ------------------------------------------------------------ handlers@@ -168,11 +203,11 @@ class Hub:} def capabilities(self) -> dict[str, Any]:- trees: list[dict[str, Any]] = []- with self._lock:- for peer in self.peers.values():- for t in peer.trees:- trees.append({**t, "peer": peer.url, "link": peer.link})+ trees = self.local_capabilities()+ if self.federation is not None:+ for remote in list(self.federation.following.values()):+ for t in remote.trees:+ trees.append({**t, "peer": remote.url, "link": remote.link, "via": "federation"})return {"name": self.name, "trees": trees} def intent(self, body: dict[str, Any]) -> tuple[int, dict[str, Any]]:@@ -194,7 +229,7 @@ class Hub:for m in matches: m.tree = tree_name # the step names the tree, not the (peer, tree) pair peer = self.holder(tree_name) or self.peers.get(peer_url)- status, answer = self.forward(peer, "/run", {"tree": tree_name, "inputs": matches[0].inputs, "depth": body.get("depth", 0), "trace": body.get("trace", True)})+ status, answer = self.forward(peer, "/run", {"tree": tree_name, "inputs": matches[0].inputs, "depth": body.get("depth", 0), "trace": body.get("trace", True), "hops": body.get("hops", [])})if isinstance(answer.get("trace"), dict) and isinstance(answer["trace"].get("steps"), list): # The hub read the sentence; the peer only ran the tree. Put the # reading where a caller expects it, at the top of the trace.@@ -217,11 +252,22 @@ class Hub:return 200, {"status": "error", "reason": "no-such-tree", "say": "No service holds the tree this snapshot came from."} return self.forward(peer, "/resume", body)- def forward(self, peer: Peer | None, path: str, body: dict[str, Any]) -> tuple[int, dict[str, Any]]:+ def forward(self, peer: Peer | Remote | None, path: str, body: dict[str, Any]) -> tuple[int, dict[str, Any]]:if peer is None: return 200, {"status": "error", "reason": "no-peer"}+ hops = list(body.get("hops") or [])+ if isinstance(peer, Remote):+ # Over federation: one hop, signed, and the request says where it has been.+ if len(hops) >= MAX_HOPS or (self.federation and self.federation.id in hops):+ return 200, {"status": "error", "reason": "hop-limit", "say": f"{peer.name or peer.url} holds this tree, but the request has already crossed a hub ({hops}); federation is {MAX_HOPS} hop.", "output": {"hops": hops, "peer": peer.url}}+ assert self.federation is not None+ body = {**body, "hops": hops + [self.federation.id]}+ headers = sign(peer.secret, self.federation.id, "POST", path, body)+ else:+ body = {k: v for k, v in body.items() if k != "hops"}+ headers = Nonetry:- status, answer = http_json("POST", f"{peer.url}{path}", body, timeout=120.0, token=self.token)+ status, answer = http_json("POST", f"{peer.url}{path}", body, timeout=120.0, token=None if headers else self.token, headers=headers)except Exception as e: peer.last_error = str(e) return 200, {"status": "error", "reason": "peer-unreachable", "say": f"{peer.name or peer.url} did not answer: {e}", "output": {"peer": peer.url}}@@ -229,6 +275,8 @@ class Hub:return 200, {"status": "error", "reason": "peer-error", "say": f"{peer.name or peer.url} answered {status}", "output": {"peer": peer.url}} peer.last_ok = time.time() answer.setdefault("peer", peer.url)+ if isinstance(peer, Remote):+ answer["via"] = [*hops, peer.url]return 200, answer def gaps(self) -> dict[str, Any]:@@ -257,13 +305,49 @@ class Hub:out.extend({**u, "peer": peer.url} for u in body.get("unanswered", [])) return {"unanswered": out}- def route(self, method: str, path: str, body: dict[str, Any] | None) -> tuple[int, Any]:+ def route(self, method: str, path: str, body: dict[str, Any] | None, headers: dict[str, str] | None = None) -> tuple[int, Any]:body = body or {}+ headers = headers or {}+ fed = self.federation+ signed = any(k.lower() == "x-pyrtan-signature" for k in headers)+ 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:+ 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 == "/health": return 200, self.health() if method == "GET" and path == "/peers": with self._lock:- return 200, {"peers": [p.to_dict() for p in self.peers.values()]}+ out: dict[str, Any] = {"peers": [p.to_dict() for p in self.peers.values()]}+ if fed is not None:+ out.update(fed.to_dict())+ return 200, out+ if fed is not None:+ if method == "GET" and path == "/actor":+ return 200, fed.actor()+ if method == "POST" and path == "/inbox":+ return fed.handle_inbox(body, headers)+ if method == "GET" and path in ("/following", "/followers"):+ return 200, {path[1:]: fed.to_dict()[path[1:]]}+ if method == "GET" and path == "/outbox":+ return 200, {"outbox": list(fed.outbox), "inbox": list(fed.inbox_log)}+ if method == "POST" and path == "/follow":+ url = body.get("url")+ if not isinstance(url, str) or not url.startswith(("http://", "https://")):+ return 400, {"status": "error", "reason": "bad-request", "say": "Send {\"url\": \"https://hub.example\"}."}+ try:+ remote = fed.follow(url)+ except ValueError as e:+ return 400, {"status": "error", "reason": "cannot-follow", "say": str(e)}+ return 200, {"status": "ok", "following": remote.to_dict()}+ if method == "POST" and path == "/unfollow":+ url = body.get("url")+ if not isinstance(url, str):+ return 400, {"status": "error", "reason": "bad-request", "say": "Send {\"url\": \"https://hub.example\"}."}+ return 200, {"status": "ok", "removed": fed.unfollow(url)}if method == "GET" and path == "/capabilities": return 200, self.capabilities() if method == "GET" and path == "/gaps":@@ -273,7 +357,10 @@ class Hub:if method == "GET" and path in ("/", "/index.html", "/console"): return 200, PAGE if method == "GET" and path == "/api":- return 200, {"pyrtan": "hub", "name": self.name, "version": __version__, "auth": "bearer" if self.token else "none", "endpoints": ["/health", "/peers", "/capabilities", "/intent", "/run", "/resume", "/gaps", "/unanswered", "/register"]}+ endpoints = ["/health", "/peers", "/capabilities", "/intent", "/run", "/resume", "/gaps", "/unanswered", "/register"]+ if fed is not None:+ endpoints += ["/actor", "/inbox", "/follow", "/unfollow", "/following", "/followers", "/outbox"]+ 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") if not isinstance(url, str) or not url.startswith(("http://", "https://")):@@ -289,14 +376,38 @@ class Hub:return 404, {"status": "error", "reason": "not-found", "say": f"No {method} {path} here."}-def _rank(peer: Peer) -> int:- return {"up": 0, "pending": 1, "stale": 2, "down": 3}[peer.link]---def serve_hub(host: str = "127.0.0.1", port: int = 8700, *, name: str = "hub", peers: list[str] | None = None, poll_every: float = 10.0, quiet: bool = False, token: str | None = None, cors: list[str] | None = None) -> tuple[Hub, Server]:+def _rank(peer: Peer | Remote) -> int:+ # A service of ours before a hub we follow; the healthier of either.+ return {"up": 0, "pending": 1, "stale": 2, "down": 3}[peer.link] + (10 if isinstance(peer, Remote) else 0)+++def serve_hub(+ host: str = "127.0.0.1",+ port: int = 8700,+ *,+ name: str = "hub",+ peers: list[str] | None = None,+ poll_every: float = 10.0,+ quiet: bool = False,+ token: str | None = None,+ cors: list[str] | None = None,+ federate: bool = False,+ advertise: str | None = None,+ federation_secret: str | None = None,+ state: str | None = None,+ open_follows: bool = True,+ follow: list[str] | None = None,+) -> tuple[Hub, Server]:hub = Hub(name=name, poll_every=poll_every, token=token)- server = Server(hub.route, host, port, quiet, token=token, cors=cors)+ 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)for url in peers or []: hub.add_peer(url)+ for url in follow or []:+ try:+ hub.federation.follow(url) # type: ignore[union-attr]+ except ValueError as e:+ print(f"pyrtan hub: cannot follow {url}: {e}", flush=True)hub.start_polling() return hub, server diff --git a/src/pyrtan/page.py b/src/pyrtan/page.py index 1ce1563..e1c2cdd 100644 --- a/src/pyrtan/page.py +++ b/src/pyrtan/page.py@@ -119,6 +119,11 @@ PAGE = r"""<!doctype html><h2>Peers</h2> <div class="card"><table id="peers"><thead><tr><th>Link</th><th>Peer</th><th>Trees</th><th>Last</th></tr></thead><tbody></tbody></table></div> <p class="mute" style="font-size:14px"><span class="link up"></span>up: answered within 30 s · <span class="link stale"></span>stale: within 90 s · <span class="link down"></span>down: longer, or never.</p>+ <div id="federation" hidden>+ <h2>Federation</h2>+ <p class="mute" style="font-size:14px;margin-top:0">This hub is <code id="fed-id"></code>. It routes to the hubs it follows; hubs that follow it receive its announcements and may ask it.</p>+ <div class="card"><table id="hubs"><thead><tr><th>Link</th><th>Hub</th><th>Relation</th><th>Trees</th></tr></thead><tbody></tbody></table></div>+ </div></section> <section id="tab-gaps" hidden>@@ -270,6 +275,20 @@ async function loadPeers() {body.appendChild(tr); } if (!(p.peers || []).length) body.innerHTML = `<tr><td colspan="4" class="mute">No peers yet. Start a service with <code>--hub ${esc(base || location.origin)}</code>.</td></tr>`;+ const fed = $("#federation");+ fed.hidden = !p.id;+ if (p.id) {+ $("#fed-id").textContent = p.id;+ const hb = $("#hubs tbody"); hb.innerHTML = "";+ for (const [rel, list] of [["follows", p.following || []], ["follower", p.followers || []]]) {+ for (const r of list) {+ const tr = document.createElement("tr");+ tr.innerHTML = `<td><span class="link ${esc(r.link)}"></span>${esc(r.link)}</td><td><code>${esc(r.name||"")}</code><div class="mute">${esc(r.url)}</div></td><td>${rel}${r.state && r.state !== "accepted" ? ` <span class="mute">(${esc(r.state)})</span>` : ""}</td><td>${(r.trees||[]).map(n => `<code>${esc(n)}</code>`).join(" ")}${r.last_error ? `<div class="failed">${esc(r.last_error)}</div>` : ""}</td>`;+ hb.appendChild(tr);+ }+ }+ if (!hb.children.length) hb.innerHTML = `<tr><td colspan="4" class="mute">Following nobody, followed by nobody. <code>pyrtan follow https://other-hub --at ${esc(base || location.origin)}</code>.</td></tr>`;+ }} async function loadGaps() { diff --git a/src/pyrtan/service.py b/src/pyrtan/service.py index 1bcc883..f26795e 100644 --- a/src/pyrtan/service.py +++ b/src/pyrtan/service.py@@ -63,17 +63,23 @@ 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"})+PUBLIC_PATHS = frozenset({"/", "/index.html", "/console", "/api", "/health", "/actor"})+# 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"})# What /health says to a caller without a token, when one is required. PUBLIC_HEALTH_KEYS = ("status", "name", "peers", "up", "stale", "down", "uptime_s")-def http_json(method: str, url: str, payload: Any = None, timeout: float = 15.0, token: str | None = None) -> tuple[int, Any]:- data = json.dumps(payload).encode("utf-8") if payload is not None else None- headers = {"Content-Type": "application/json", "Accept": "application/json"}+def http_json(method: str, url: str, payload: Any = None, timeout: float = 15.0, token: str | None = None, headers: dict[str, str] | None = None) -> tuple[int, Any]:+ # Canonical JSON on the wire, so a signature over the body (federation)+ # can be checked from the parsed body on the other side.+ data = json.dumps(payload, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode("utf-8") if payload is not None else None+ hdrs = {"Content-Type": "application/json", "Accept": "application/json"}if token:- headers["Authorization"] = f"Bearer {token}"- req = urllib.request.Request(url, data=data, method=method, headers=headers)+ hdrs["Authorization"] = f"Bearer {token}"+ hdrs.update(headers or {})+ req = urllib.request.Request(url, data=data, method=method, headers=hdrs)try: with urllib.request.urlopen(req, timeout=timeout) as resp: return resp.status, json.loads(resp.read().decode("utf-8") or "null")@@ -205,7 +211,7 @@ class Service:# ------------------------------------------------------------ routing- def route(self, method: str, path: str, body: dict[str, Any] | None) -> tuple[int, Any]:+ def route(self, method: str, path: str, body: dict[str, Any] | None, headers: dict[str, str] | None = None) -> tuple[int, Any]:body = body or {} if method == "GET" and path == "/health": return 200, self.health()@@ -228,7 +234,7 @@ class Service:return 404, {"status": "error", "reason": "not-found", "say": f"No {method} {path} here."}-Route = Callable[[str, str, dict[str, Any] | None], tuple[int, Any]]+Route = Callable[..., tuple[int, Any]] # (method, path, body, headers) -> (status, payload)def bearer_ok(header: str | None, token: str) -> bool:@@ -239,7 +245,7 @@ def bearer_ok(header: str | None, token: str) -> bool:return scheme.lower() == "bearer" and hmac.compare_digest(value.strip().encode("utf-8"), token.encode("utf-8"))-def make_handler(route: Route, quiet: bool = False, token: str | None = None, cors: list[str] | None = None):+def make_handler(route: Route, quiet: bool = False, token: str | None = None, cors: list[str] | None = None, federated: bool = False):token = token or None origins = [o.rstrip("/") for o in (cors or []) if o] any_origin = "*" in origins@@ -281,6 +287,8 @@ 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"):+ return True # the route verifies the signatureself._send(401, {"status": "error", "reason": "unauthorised", "say": "This Pyrtan needs a token: send Authorization: Bearer <token>."}, extra={"WWW-Authenticate": 'Bearer realm="pyrtan"'}) return False@@ -316,7 +324,7 @@ def make_handler(route: Route, quiet: bool = False, token: str | None = None, copath = self.path.split("?", 1)[0] if not self._authorised(path): return- status, payload = route("GET", path, None)+ status, payload = route("GET", path, None, dict(self.headers))self._send(status, self._public_view(path, payload)) def do_POST(self) -> None:@@ -336,7 +344,7 @@ def make_handler(route: Route, quiet: bool = False, token: str | None = None, coif not isinstance(body, dict): self._send(400, {"status": "error", "reason": "bad-json", "say": "The body must be a JSON object."}) return- status, payload = route("POST", path, body)+ status, payload = route("POST", path, body, dict(self.headers))self._send(status, payload) return Handler@@ -348,8 +356,8 @@ class Server:``token`` requires ``Authorization: Bearer`` on everything but the public paths; ``cors`` lists the browser origins allowed to call (``*`` for any)."""- def __init__(self, route: Route, host: str = "127.0.0.1", port: int = 0, quiet: bool = False, *, token: str | None = None, cors: list[str] | None = None) -> None:- self.httpd = ThreadingHTTPServer((host, port), make_handler(route, quiet, token, cors))+ def __init__(self, route: Route, host: str = "127.0.0.1", port: int = 0, quiet: bool = False, *, token: str | None = None, cors: list[str] | None = None, federated: bool = False) -> None:+ self.httpd = ThreadingHTTPServer((host, port), make_handler(route, quiet, token, cors, federated))self.httpd.daemon_threads = True self.thread: threading.Thread | None = None diff --git a/src/pyrtan/transport.py b/src/pyrtan/transport.py index b55096e..1768f44 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.2 (+https://github.com/b-i-t-8/pyrtan)", egress: str = "any") -> None:+ def __init__(self, user_agent: str = "pyrtan/0.3 (+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 new file mode 100644 index 0000000..f36cef7 --- /dev/null +++ b/tests/test_federation.py@@ -0,0 +1,265 @@+"""Two hubs, each with its own service, following each other over real+sockets: who may ask whom, what is announced, and what is refused."""++import json+import time++import pytest++from pyrtan import Registry, RecordedTransport+from pyrtan.cli import main+from pyrtan.federation import Federation, Remote, sign, verify+from pyrtan.hub import Hub+from pyrtan.service import Server, Service, http_json+from conftest import EXAMPLES, FIXTURES++ECHO = """+name = "{name}"+description = "echoes"+start = "say"++[[triggers]]+pattern = "{word} {{text}}"++[inputs.text]+type = "str"++[nodes.say]+kind = "done"+say = "{name}: {{{{ inputs.text }}}}"+output = {{ text = "{{{{ inputs.text }}}}" }}+"""+++class Swarm:+ """A hub with a token, federated, holding one service with the given trees."""++ def __init__(self, name: str, tree_dir, token: str | None = None, **fed):+ self.name = name+ self.token = token+ self.hub = Hub(name=name, poll_every=60, token=token)+ self.hub_srv = Server(self.hub.route, quiet=True, token=token, federated=True).start()+ self.fed = self.hub.federate(self.hub_srv.url, **fed)+ self.svc = Service(Registry.from_paths([tree_dir]), name=f"{name}-svc", transport=RecordedTransport.from_file(FIXTURES / "examples.json"), token=token, hub=self.hub_srv.url)+ self.svc_srv = Server(self.svc.route, quiet=True, token=token).start()+ assert self.svc.register(self.svc_srv.url)++ @property+ def url(self):+ return self.hub_srv.url++ def get(self, path):+ return http_json("GET", f"{self.url}{path}", token=self.token)++ def post(self, path, body):+ return http_json("POST", f"{self.url}{path}", body, token=self.token)++ def stop(self):+ self.hub.stop()+ self.hub_srv.stop()+ self.svc_srv.stop()+++@pytest.fixture+def two(tmp_path):+ (tmp_path / "a").mkdir()+ (tmp_path / "b").mkdir()+ (tmp_path / "a" / "alpha.toml").write_text(ECHO.format(name="alpha", word="alpha"))+ (tmp_path / "b" / "beta.toml").write_text(ECHO.format(name="beta", word="beta"))+ a = Swarm("a", tmp_path / "a", token="token-a")+ b = Swarm("b", tmp_path / "b", token="token-b")+ yield a, b+ a.stop()+ b.stop()+++def test_follow_accept_announce_and_route(two):+ a, b = two+ status, actor = http_json("GET", f"{b.url}/actor") # public, no token+ assert status == 200 and actor["type"] == "Hub" and actor["inbox"] == f"{b.url}/inbox" and actor["follows"] == "open"++ status, out = a.post("/follow", {"url": b.url})+ assert status == 200 and out["following"]["url"] == b.url+ # The handshake is synchronous: B fetched A's actor, accepted, announced.+ assert a.fed.following[b.url].state == "accepted" and a.fed.following[b.url].link == "up"+ assert b.fed.followers[a.url].state == "accepted"+ assert [t["name"] for t in a.fed.following[b.url].trees] == ["beta"]++ _, caps = a.get("/capabilities")+ beta = next(t for t in caps["trees"] if t["name"] == "beta")+ assert beta["via"] == "federation" and beta["peer"] == b.url and beta["link"] == "up"+ _, peers = a.get("/peers")+ assert peers["id"] == a.url and peers["following"][0]["url"] == b.url and peers["followers"] == []+ _, peers = b.get("/peers")+ assert peers["followers"][0]["url"] == a.url and peers["following"] == []++ # A asks for B's tree: routed over federation, signed, one hop, and the answer says so.+ _, out = a.post("/intent", {"text": "beta hello"})+ assert out["status"] == "done" and out["say"] == "beta: hello"+ assert out["peer"] == b.svc_srv.url and out["via"] == [b.url]+ assert out["trace"]["steps"][0]["kind"] == "intent"+ _, out = a.post("/run", {"tree": "beta", "inputs": {"text": "x"}})+ assert out["status"] == "done" and out["via"] == [b.url]+ # Its own service still comes first for its own trees.+ _, out = a.post("/intent", {"text": "alpha hi"})+ assert out["status"] == "done" and "via" not in out++ # Being followed grants nothing: B does not route to A.+ _, caps = b.get("/capabilities")+ assert [t["name"] for t in caps["trees"]] == ["beta"]+ _, out = b.post("/intent", {"text": "alpha hi"})+ assert out["reason"] == "no-match"++ _, box = a.get("/outbox")+ assert [x["type"] for x in box["outbox"]] == ["Follow"] and box["outbox"][0]["delivered"]+ assert [x["type"] for x in box["inbox"]] == ["Accept", "Announce"]+++def test_signed_requests_replace_the_token_only_for_hubs_we_know(two):+ a, b = two+ a.post("/follow", {"url": b.url})+ secret = a.fed.following[b.url].secret+ body = {"tree": "beta", "inputs": {"text": "x"}}+ # No token, no signature: refused.+ assert http_json("POST", f"{b.url}/run", body)[0] == 401+ # Signed by A with the pair secret: served.+ status, out = http_json("POST", f"{b.url}/run", body, headers=sign(secret, a.url, "POST", "/run", body))+ assert status == 200 and out["status"] == "done"+ # Wrong secret, wrong actor, stale date: refused.+ assert http_json("POST", f"{b.url}/run", body, headers=sign("nope", a.url, "POST", "/run", body))[0] == 401+ assert http_json("POST", f"{b.url}/run", body, headers=sign(secret, "http://stranger", "POST", "/run", body))[0] == 401+ assert http_json("POST", f"{b.url}/run", body, headers=sign(secret, a.url, "POST", "/run", body, date=time.time() - 3600))[0] == 401+ # A signature does not open the operator's paths.+ assert http_json("GET", f"{b.url}/peers", headers=sign(secret, a.url, "GET", "/peers", None))[0] == 401+ assert http_json("POST", f"{b.url}/follow", {"url": a.url}, headers=sign(secret, a.url, "POST", "/follow", {"url": a.url}))[0] == 401+ # The inbox checks the activity's actor against the signature's.+ announce = {"type": "Announce", "actor": a.url, "object": {"type": "Capabilities", "trees": []}}+ status, out = http_json("POST", f"{b.url}/inbox", announce, headers=sign(secret, a.url, "POST", "/inbox", announce))+ assert status == 403 and out["reason"] == "unknown-actor" # B does not follow A, so A cannot announce to it+ bad = {"type": "Announce", "actor": b.url, "object": {"type": "Capabilities", "trees": []}}+ assert http_json("POST", f"{a.url}/inbox", bad, headers=sign("nope", b.url, "POST", "/inbox", bad))[0] == 401+ assert http_json("POST", f"{a.url}/inbox", {"type": "Dance", "actor": b.url}, headers=sign(secret, b.url, "POST", "/inbox", {"type": "Dance", "actor": b.url}))[0] == 400+++def test_federation_is_one_hop(two, tmp_path):+ a, b = two+ (tmp_path / "c").mkdir()+ (tmp_path / "c" / "gamma.toml").write_text(ECHO.format(name="gamma", word="gamma"))+ c = Swarm("c", tmp_path / "c")+ try:+ c.post("/follow", {"url": a.url})+ a.post("/follow", {"url": b.url})+ # C knows A's own trees, not what A learned from B.+ _, caps = c.get("/capabilities")+ assert {t["name"] for t in caps["trees"]} == {"gamma", "alpha"}+ assert c.post("/intent", {"text": "beta hello"})[1]["reason"] == "no-match"+ assert c.post("/run", {"tree": "beta", "inputs": {"text": "x"}})[1]["reason"] == "no-such-tree"+ # A request that already crossed a hub is not forwarded again.+ secret = c.fed.following[a.url].secret+ body = {"tree": "beta", "inputs": {"text": "x"}, "hops": [c.url]}+ status, out = http_json("POST", f"{a.url}/run", body, headers=sign(secret, c.url, "POST", "/run", body))+ assert status == 200 and out["reason"] == "hop-limit"+ # A mutual follow does not loop: a sentence nobody answers is simply unanswered.+ b.post("/follow", {"url": a.url})+ assert a.post("/intent", {"text": "nobody answers this"})[1]["reason"] == "no-match"+ assert b.post("/intent", {"text": "alpha hi"})[1]["status"] == "done"+ finally:+ c.stop()+++def test_follow_policy_secret_and_closed(tmp_path):+ (tmp_path / "t").mkdir()+ (tmp_path / "t" / "alpha.toml").write_text(ECHO.format(name="alpha", word="alpha"))+ guarded = Swarm("guarded", tmp_path / "t", secret="federation-secret")+ stranger = Swarm("stranger", tmp_path / "t")+ insider = Swarm("insider", tmp_path / "t", secret="federation-secret")+ closed = Swarm("closed", tmp_path / "t", open_follows=False)+ try:+ assert http_json("GET", f"{guarded.url}/actor")[1]["follows"] == "secret"+ stranger.post("/follow", {"url": guarded.url})+ assert stranger.fed.following[guarded.url].state == "pending"+ assert "federation secret" in stranger.fed.following[guarded.url].last_error+ assert stranger.url not in guarded.fed.followers+ insider.post("/follow", {"url": guarded.url})+ assert insider.fed.following[guarded.url].state == "accepted"+ stranger.post("/follow", {"url": closed.url})+ assert stranger.fed.following[closed.url].state == "pending" and "does not take follows" in stranger.fed.following[closed.url].last_error+ # A closed hub can still follow.+ closed.post("/follow", {"url": stranger.url})+ assert closed.fed.following[stranger.url].state == "accepted"+ status, out = stranger.post("/follow", {"url": stranger.url})+ assert status == 400 and "itself" in out["say"]+ status, out = stranger.post("/follow", {"url": stranger.svc_srv.url})+ assert status == 400 and "does not answer /actor" in out["say"]+ finally:+ for s in (guarded, stranger, insider, closed):+ s.stop()+++def test_unfollow_announce_on_change_and_state(two, tmp_path):+ a, b = two+ state = tmp_path / "a-state.json"+ a.fed.state_path = state+ a.post("/follow", {"url": b.url})+ assert json.loads(state.read_text())["following"][0]["url"] == b.url+ # B gains a tree; its next poll notices and announces; A learns without polling B.+ (tmp_path / "b" / "beta2.toml").write_text(ECHO.format(name="beta-two", word="betatwo"))+ b.hub.refresh_all()+ assert {t["name"] for t in a.fed.following[b.url].trees} == {"beta", "beta-two"}+ assert b.fed.announce() == 0 # nothing changed since: no delivery+ # A fresh federation over the same file remembers the pair.+ again = Federation(a.url, state=state)+ assert again.following[b.url].secret == a.fed.following[b.url].secret and again.following[b.url].state == "accepted"+ # Unfollow: both sides forget, and A stops routing there.+ status, out = a.post("/unfollow", {"url": b.url})+ assert status == 200 and out["removed"] is True+ assert b.url not in a.fed.following and a.url not in b.fed.followers+ assert a.post("/intent", {"text": "beta hello"})[1]["reason"] == "no-match"+ assert a.post("/unfollow", {"url": b.url})[1]["removed"] is False+ assert json.loads(state.read_text()) == {"following": [], "followers": []}+++def test_unfederated_hub_has_no_inbox(tmp_path):+ hub = Hub()+ srv = Server(hub.route, quiet=True).start()+ try:+ assert http_json("GET", f"{srv.url}/actor")[0] == 404+ assert http_json("POST", f"{srv.url}/follow", {"url": "http://x"})[0] == 404+ assert http_json("GET", f"{srv.url}/api")[1]["federation"] is None+ finally:+ srv.stop()+++def test_cli_follow_peers_unfollow(two, capsys):+ a, b = two++ def run(*argv):+ code = main(list(argv))+ out = capsys.readouterr()+ return code, out.out, out.err++ code, out, _ = run("follow", b.url, "--at", a.url, "--token", "token-a")+ assert code == 0 and "now follows" in out and "accepted" in out+ code, out, _ = run("peers", "--at", a.url, "--token", "token-a")+ assert code == 0 and "follows" in out and "beta" in out and "a-svc" in out+ code, out, _ = run("peers", "--at", b.url, "--token", "token-b")+ assert "follower" in out+ assert run("follow", b.url, "--at", a.url)[0] == 3 # no token+ code, out, _ = run("unfollow", b.url, "--at", a.url, "--token", "token-a")+ assert code == 0 and "no longer follows" in out+ code, _, err = run("follow", "http://127.0.0.1:9", "--at", a.url, "--token", "token-a")+ assert code == 1 and "does not answer /actor" in err+++def test_signing_primitives():+ headers = sign("s", "http://a", "POST", "/inbox", {"b": 1, "a": [1, 2]})+ assert verify("s", "POST", "/inbox", {"a": [1, 2], "b": 1}, headers) is None # key order does not matter+ assert verify("s", "POST", "/inbox", {"a": [1, 2], "b": 2}, headers) == "bad signature"+ assert verify("t", "POST", "/inbox", {"a": [1, 2], "b": 1}, headers) == "bad signature"+ assert verify("s", "POST", "/other", {"a": [1, 2], "b": 1}, headers) == "bad signature"+ assert verify("s", "POST", "/inbox", {"a": [1, 2], "b": 1}, {}) == "no signature"+ assert verify("s", "POST", "/inbox", {"a": [1, 2], "b": 1}, headers, now=time.time() + 1000) == "signature too old"+ assert Remote("http://b", "http://b/inbox", "s").link == "pending"+ assert Remote("http://b", "http://b/inbox", "s", state="accepted").link == "pending" # accepted, not heard from yet+ assert Remote("http://b", "http://b/inbox", "s", state="accepted", last_error="x").link == "down"+ assert Remote("http://b", "http://b/inbox", "s", state="rejected").link == "down"diff --git a/uv.lock b/uv.lock index 4beb188..6c6c2e8 100644 --- a/uv.lock +++ b/uv.lock@@ -147,7 +147,7 @@ wheels = [[[package]] name = "pyrtan"-version = "0.2.0"+version = "0.3.0"source = { editable = "." } [package.optional-dependencies]