Key backups, and names any network can run on any git host
Claude · 2026-10-28 · 38f14f24c89fe317677cbc42d1fa61e73cfbded0 · parents d610257
identity.py: a root key encrypted with a passphrase (PBKDF2-SHA256, 600k rounds, then AES-256-GCM over the seed with the identifier as associated data), its identity document beside it; releases (a name given up, signed by its key) and handovers (a name moved to a new key, signed by the old one first). pyrtan id backup, restore, release, and id name --handover-from.
names.py: gate_key, the rule for names owned by their keys, which needs only the two trees and signatures: a new name is open to whoever signs it, it moves only with a handover, it is given up only by a release, and an identity document never goes back to an older one or away while its name stands. pyrtan names init writes a names repository for a domain (README, names/, pyrtan-names.json, and for GitHub a pull_request_target gate that compares the pull request with its merge base and Pages, or for GitLab a .gitlab-ci.yml); pyrtan names gate runs the rule; pyrtan id name --into works on such a repository. pyrtan hub --names DIR serves a network's names, and their documents, from its own hubs.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_013rD7jgoLgxA3qCSMVWdoDv
Changes
src/pyrtan/cli.py | 147 ++++++++++++++++++++++++++++-- src/pyrtan/hub.py | 48 +++++++++- src/pyrtan/identity.py | 146 +++++++++++++++++++++++++++++- src/pyrtan/names.py | 341 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++------- tests/test_identity.py | 63 +++++++++++++ tests/test_names.py | 163 +++++++++++++++++++++++++++++++++ 6 files changed, 863 insertions(+), 45 deletions(-)
diff --git a/src/pyrtan/cli.py b/src/pyrtan/cli.py index 957dbbb..5369629 100644 --- a/src/pyrtan/cli.py +++ b/src/pyrtan/cli.py@@ -29,6 +29,7 @@pyrtan id new|show|check|device|revoke|route|name your identity: a key, any domain pyrtan id resolve NAME@DOMAIN what identity a name is, pinned the first time pyrtan allow WHO --may ask,run --at HUB grant an identity abilities on your hub; allowed, disallow+ pyrtan names init DIR --domain D | gate a network's own names, in any git repository``serve`` and ``hub`` take ``--token`` (or ``$PYRTAN_TOKEN``) to require a bearer token, ``--cors ORIGIN`` to let a browser elsewhere call them, and@@ -177,6 +178,8 @@ def main(argv: list[str] | None = None) -> int: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("--chat", action="store_true", default=os.environ.get("PYRTAN_CHAT", "").lower() in ("1", "true", "yes"), help="keep a lobby at /chat where people meet a random stranger here or at a hub this one follows, or a bot made of the trees it routes to (default $PYRTAN_CHAT)") p.add_argument("--allow", action="append", default=[a for a in os.environ.get("PYRTAN_ALLOW", "").split() if a], metavar="WHO=ABILITIES", help="grant an identity (pyr:… or name@domain, resolved and pinned now) abilities here: ask, chat, run, operate (repeatable; default $PYRTAN_ALLOW, space-separated)")+ p.add_argument("--names", metavar="DIR", default=os.environ.get("PYRTAN_NAMES"), help="serve the name claims in DIR (a checkout of a names repository, `pyrtan names init`) at /.well-known/pyrtan/, as this network's name provider (default $PYRTAN_NAMES)")+ p.add_argument("--names-domain", metavar="DOMAIN", default=os.environ.get("PYRTAN_NAMES_DOMAIN"), help="the domain the names are for (default $PYRTAN_NAMES_DOMAIN, else this hub's own host); the hub must be reachable at https://DOMAIN")p.add_argument("--grants", metavar="FILE", default=os.environ.get("PYRTAN_GRANTS"), help="keep the grants in this file (default $PYRTAN_GRANTS, else grants.json beside --state, else in memory)") p.add_argument("--chat-public", action="store_true", default=os.environ.get("PYRTAN_CHAT_PUBLIC", "").lower() in ("1", "true", "yes"), help="open the lobby to callers without the token (implies --chat): they meet people only, never a bot, and everything else still needs the token (default $PYRTAN_CHAT_PUBLIC)") p.add_argument("--chat-typing", action=argparse.BooleanOptionalAction, default=os.environ.get("PYRTAN_CHAT_TYPING", "").lower() not in ("0", "false", "no", "off"), help="the lobby's typing indicator: \"Stranger is typing…\" for anyone who keeps theirs on (default on; $PYRTAN_CHAT_TYPING=0 turns it off)")@@ -279,6 +282,18 @@ def main(argv: list[str] | None = None) -> int:p.add_argument("--into", required=True, metavar="CHECKOUT", help="a checkout of github.com/b-i-t-8/pyrtan (or any repository with a catalog/ directory)") p.add_argument("--owner", action="append", default=[], metavar="LOGIN", help="GitHub login(s) that own the namespace; needed when the namespace is new")+ p = sub.add_parser("names", help="a network's own names, in any git repository: make one, or run its gate")+ nms = p.add_subparsers(dest="names_command", required=True)+ q = nms.add_parser("init", help="write a names repository for DOMAIN: README, names/, and the gate and Pages for its git host")+ q.add_argument("dir", type=Path)+ q.add_argument("--domain", required=True, help="the domain the names are for: name@DOMAIN, served at https://DOMAIN/.well-known/pyrtan/")+ q.add_argument("--host", choices=["github", "gitlab", "plain"], default="github", help="github: Actions (pull_request_target) and Pages; gitlab: .gitlab-ci.yml and Pages; plain: neither (default github)")+ q.add_argument("--pyrtan-ref", default="main", metavar="REF", help="the pyrtan the gate installs (a branch, tag or commit; pin a commit you have read; default main)")+ q = nms.add_parser("gate", help="judge a change to a directory of names owned by their keys: HEAD against BASE")+ q.add_argument("head", type=Path, help="the names directory as the change leaves it")+ q.add_argument("--base", type=Path, help="the names directory the change starts from (its merge base); omit for a first commit")+ q.add_argument("--domain", required=True)+p = sub.add_parser("allow", help="grant an identity abilities on your hub: ask, chat, run, operate") p.add_argument("who", metavar="pyr:…|NAME@DOMAIN", help="an identity, or a name your hub resolves and pins now") p.add_argument("--may", required=True, metavar="ABILITIES", help="comma-separated: ask (the routed paths), chat (the lobby), run (be a trusted member), operate (what the token allows)")@@ -323,13 +338,23 @@ def main(argv: list[str] | None = None) -> int:q.add_argument("--github", type=int, metavar="USER_ID", help="your numeric GitHub user id, for a name at pyrtan.com (with --into, looked up from the name when omitted)") where = q.add_mutually_exclusive_group() where.add_argument("--out", help="write the claim here (default: stdout)")- where.add_argument("--into", type=Path, metavar="CHECKOUT", help="put the claim and your identity document into a checkout of this repository's site/names/, ready for a pull request")+ where.add_argument("--into", type=Path, metavar="CHECKOUT", help="put the claim and your identity document into a checkout of a network's names repository (pyrtan names init), or of this repository's site/names/, ready for a pull request")+ q.add_argument("--handover-from", type=Path, metavar="KEY-FILE", help="the key that holds the name now (its root.key): it signs the handover that moves the name to you")+ q = ids.add_parser("release", help="sign that you give up a name, to replace your claim with in the names repository")+ q.add_argument("name", metavar="NAME@DOMAIN")+ q.add_argument("--out", help="write the release here (default: stdout)")q = ids.add_parser("resolve", help="what identity a name@domain is, pinned the first time") q.add_argument("name", metavar="NAME@DOMAIN") q.add_argument("--repin", action="store_true", help="accept a different identity than the one pinned") q.add_argument("--json", action="store_true", help="print the claim, and the identity document if the domain serves one") q = ids.add_parser("pins", help="the names you have resolved, and the identity each is pinned to") q.add_argument("--forget", metavar="NAME@DOMAIN", help="drop a pin, so the next resolve takes what the domain says")+ q = ids.add_parser("backup", help="write your key, encrypted with a passphrase, and your identity document, to keep somewhere only you can read")+ q.add_argument("--out", required=True, help="the file to write (a private place: your disk, your own private repository; never a public one)")+ q.add_argument("--passphrase-env", metavar="VAR", help="read the passphrase from this environment variable instead of asking")+ q = ids.add_parser("restore", help="make this machine your identity again, from a backup (the console's or `pyrtan id backup`'s)")+ q.add_argument("file", help="the backup file")+ q.add_argument("--passphrase-env", metavar="VAR", help="read the passphrase from this environment variable instead of asking")q = ids.add_parser("push", help="give a hub your newest identity document, so it knows your devices and revocations now") q.add_argument("--at", default=os.environ.get("PYRTAN_AT", "http://127.0.0.1:8700"), help="the hub (default $PYRTAN_AT or the local hub)") q = ids.add_parser("wellknown", help="build /.well-known/pyrtan/ from a directory of name claims, for any domain to serve")@@ -346,7 +371,7 @@ def main(argv: list[str] | None = None) -> int:"search": cmd_search, "pull": cmd_pull, "index": cmd_index, "publish": cmd_publish, "console": cmd_console, "follow": cmd_follow, "join": cmd_join, "network": cmd_network, "unfollow": cmd_unfollow, "peers": cmd_peers, "directory": cmd_directory, "reports": cmd_reports, "keep": cmd_keep, "kept": cmd_kept, "fetch": cmd_fetch, "id": cmd_id,- "allow": cmd_allow, "allowed": cmd_allowed, "disallow": cmd_disallow,+ "allow": cmd_allow, "allowed": cmd_allowed, "disallow": cmd_disallow, "names": cmd_names,} return commands[args.command](args)@@ -892,7 +917,7 @@ def cmd_hub(args: argparse.Namespace) -> int:state=args.state, open_follows=not args.closed, follow=follow, join=join, listed=args.listed, trees=[Path(t) for t in args.trees], handout=[Path(t) for t in args.handout], env=_kv(args.env), gaps=args.gaps, egress=args.egress, store=args.store, replicas=args.replicas, report_gaps=args.report_gaps, unanswered=args.unanswered, watch=args.watch, chat=args.chat or args.chat_public, chat_typing=args.chat_typing, chat_public=args.chat_public,- allow=args.allow, grants=args.grants,+ allow=args.allow, grants=args.grants, names=args.names, names_domain=args.names_domain,) except IdentityError as e: print(f"pyrtan hub: {e}", file=sys.stderr)@@ -1383,6 +1408,7 @@ def cmd_id(args: argparse.Namespace) -> int:"new": _id_new, "show": _id_show, "check": _id_check, "device": _id_device, "revoke": _id_revoke, "route": _id_route, "name": _id_name, "resolve": _id_resolve, "pins": _id_pins, "wellknown": _id_wellknown, "push": _id_push,+ "backup": _id_backup, "restore": _id_restore, "release": _id_release,}[args.id_command](args, directory) except IdentityError as e: print(f"pyrtan id: {e}", file=sys.stderr)@@ -1470,6 +1496,10 @@ def _id_check(args: argparse.Namespace, directory: Path) -> int:print(f"{thing}: the identity document of {doc['id']}, signed by it; {len(doc['devices'])} device(s), {len(doc['revoked'])} revoked") elif kind == "device": print(f"{thing}: {doc['device']} may {','.join(doc['may']) or 'nothing'} for {doc['root']}, until {time.strftime('%Y-%m-%d', time.localtime(doc['expires']))}")+ elif kind == "release":+ print(f"{thing}: {doc['id']} gives up {doc['name']}@{doc['domain']}, signed by it")+ elif kind == "backup":+ print(f"{thing}: a key backup of {doc['id']}, {doc['kdf']['iterations']} PBKDF2 rounds" + (", with its identity document" if doc.get("document") else "") + "; `pyrtan id restore` opens it")else: print(f"{thing}: {doc['id']} claims {doc['name']}@{doc['domain']}, signed by it") return 0@@ -1511,23 +1541,27 @@ def _id_route(args: argparse.Namespace, directory: Path) -> int:def _id_name(args: argparse.Namespace, directory: Path) -> int: from . import identity, names key, doc = _id_load(directory)+ network = names.repo_domain(args.into) if args.into else Nonegithub = args.github- if args.into and github is None:+ if args.into and network is None and github is None:local = identity.parse(args.name).name github = names.github_id(local) print(f"GitHub says {local} is user {github}", file=sys.stderr) extra = {"github": github} if github is not None else {}- claim = identity.make_name(key, args.name, **extra)+ old = identity.Key.load(args.handover_from) if args.handover_from else None+ claim = identity.make_name(key, args.name, handover_from=old, **extra)text = json.dumps(claim, indent=2, ensure_ascii=False) + "\n" if args.into: written = names.place(args.into, claim, doc)- problems = names.gate(args.into)+ problems = names.gate_key(args.into / "names", None, network) if network else names.gate(args.into)for p in problems: print(f"gate: {p}", file=sys.stderr) if problems: return 1 print("wrote " + " and ".join(str(w) for w in written))- print(f"now: git checkout -b names/{claim['name']} && git add {names.NAMES_DIR} && git commit && git push -u origin HEAD, and open a pull request from the GitHub account {claim['name']}")+ where = "names" if network else names.NAMES_DIR+ print(f"now: git checkout -b names/{claim['name']} && git add {where} && git commit && git push -u origin HEAD, and open a pull request"+ + ("" if network else f" from the GitHub account {claim['name']}"))return 0 if args.out: Path(args.out).write_text(text, encoding="utf-8")@@ -1537,6 +1571,19 @@ def _id_name(args: argparse.Namespace, directory: Path) -> int:return 0+def _id_release(args: argparse.Namespace, directory: Path) -> int:+ from . import identity+ key, _ = _id_load(directory)+ release = identity.make_release(key, args.name)+ text = json.dumps(release, indent=2, ensure_ascii=False) + "\n"+ if args.out:+ Path(args.out).write_text(text, encoding="utf-8")+ print(f"wrote {args.out}: {key.id} gives up {release['name']}@{release['domain']}; put it in place of names/{release['name']}.json")+ else:+ sys.stdout.write(text)+ return 0++def _id_resolve(args: argparse.Namespace, directory: Path) -> int: from . import identity, names claim = names.resolve(args.name)@@ -1588,6 +1635,64 @@ def _id_wellknown(args: argparse.Namespace, directory: Path) -> int:return 1 if problems else 0+def _passphrase(args: argparse.Namespace, *, twice: bool) -> str:+ import getpass+ if args.passphrase_env:+ value = os.environ.get(args.passphrase_env)+ if not value:+ raise IdentityError(f"${args.passphrase_env} is empty")+ return value+ first = getpass.getpass("passphrase: ")+ if twice and getpass.getpass("again: ") != first:+ raise IdentityError("the two passphrases differ")+ return first+++def _id_backup(args: argparse.Namespace, directory: Path) -> int:+ from . import identity+ key, doc = _id_load(directory)+ out = Path(args.out)+ if out.exists():+ raise IdentityError(f"{out} exists; a backup is not overwritten")+ backup = identity.make_backup(key, _passphrase(args, twice=True), document=doc)+ fd = os.open(out, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)+ with os.fdopen(fd, "w", encoding="utf-8") as f:+ f.write(json.dumps(backup, indent=2) + "\n")+ print(f"wrote {out}: {key.id}, encrypted with your passphrase, and your identity document")+ print("keep it where only you can read it (your disk, your own private repository), never in a public one: anyone with a copy can guess at the passphrase forever")+ return 0+++def _id_restore(args: argparse.Namespace, directory: Path) -> int:+ from . import identity+ try:+ backup = json.loads(Path(args.file).read_text(encoding="utf-8"))+ except (OSError, json.JSONDecodeError) as e:+ raise IdentityError(f"{args.file}: {e}") from None+ problem = identity.backup_problem(backup)+ if problem:+ raise IdentityError(f"{args.file}: {problem}")+ key_file = directory / "root.key"+ if key_file.exists():+ held = identity.Key.load(key_file)+ if held.id != backup["id"].lower():+ raise IdentityError(f"{directory} already holds {held.id}; move it away first, knowing that it is an identity")+ key = identity.open_backup(backup, _passphrase(args, twice=False))+ doc = backup.get("document") or identity.make_identity(key)+ current = None+ if (directory / "identity.json").exists():+ try:+ current = json.loads((directory / "identity.json").read_text(encoding="utf-8"))+ except (OSError, json.JSONDecodeError):+ current = None+ doc = identity.newest([doc, current], ident=key.id) or doc+ if not key_file.exists():+ key.save(key_file)+ _id_save(directory, doc)+ print(f"you are {key.id} again, in {directory}" + ("" if backup.get("document") else "; the backup held no identity document, so a new one lists no devices"))+ return 0++def _id_push(args: argparse.Namespace, directory: Path) -> int: _, doc = _id_load(directory) at = args.at.rstrip("/")@@ -1603,6 +1708,34 @@ def _id_push(args: argparse.Namespace, directory: Path) -> int:return 0+# ---------------------------------------------------------------- a network's names+++def cmd_names(args: argparse.Namespace) -> int:+ from . import identity, names+ try:+ if args.names_command == "init":+ written, kept = names.init_repo(args.dir, args.domain, host=args.host, pyrtan_ref=args.pyrtan_ref)+ for path in written:+ print(f"wrote {path}")+ for path in kept:+ print(f"kept {path} (already there)")+ print(f"next: commit it, push it to your git host, point {args.domain} at its Pages (or at a hub started --names {args.dir / 'names'}),"+ f" and claim the first name: pyrtan id name <you>@{args.domain} --into {args.dir}")+ return 0+ domain = identity.check_host(args.domain.lower())+ if not args.head.is_dir():+ raise IdentityError(f"{args.head} is not a directory")+ problems = names.gate_key(args.head, args.base, domain, label=args.head.name or "names")+ except IdentityError as e:+ print(f"pyrtan names: {e}", file=sys.stderr)+ return 2+ for p in problems:+ print(f"gate: {p}")+ print(f"gate: {'refused' if problems else 'open'} ({domain}, names owned by their keys)")+ return 1 if problems else 0++# ---------------------------------------------------------------- grants diff --git a/src/pyrtan/hub.py b/src/pyrtan/hub.py index a71034f..6bbe566 100644 --- a/src/pyrtan/hub.py +++ b/src/pyrtan/hub.py@@ -49,6 +49,10 @@ may no longer be. ``GET``/``POST /allow`` and ``POST /disallow`` (operator)keep the grants; ``POST /identity`` takes a granted root's newer identity document from anyone, since it is signed; ``GET /.well-known/pyrtan/id/<56>.json`` serves the documents the hub holds.+With ``--names DIR`` a hub is also its network's name provider: it serves+the claims in DIR (``pyrtan names init`` makes a repository of them) at+``/.well-known/pyrtan/name/<name>.json``, ``names.json``, and their+identity documents, for its own domain.""" from __future__ import annotations@@ -143,8 +147,11 @@ class _HubResolver:class Hub:- def __init__(self, *, name: str = "hub", poll_every: float = 10.0, token: str | None = None, local: Service | None = None, handout: Registry | None = None, unanswered: str | Path | None = None, chat: bool = False, chat_typing: bool = True, chat_public: bool = False, grants: Grants | None = None, audience: str | None = None) -> None:+ def __init__(self, *, name: str = "hub", poll_every: float = 10.0, token: str | None = None, local: Service | None = None, handout: Registry | None = None, unanswered: str | Path | None = None, chat: bool = False, chat_typing: bool = True, chat_public: bool = False, grants: Grants | None = None, audience: str | None = None, names: str | Path | None = None, names_domain: str | None = None) -> None:self.name = name+ self.names_dir = Path(names) if names else None+ self._names_domain = names_domain+ self._names_cache: tuple[tuple, Any] | None = Noneself.grants = grants if grants is not None else Grants() self.grants.on_change = self._grant_changed self._audience = audience@@ -189,6 +196,27 @@ class Hub:"""The id a key-signed request to this hub must be signed for.""" return self.federation.id if self.federation is not None else self._audience+ @property+ def names_domain(self) -> str | None:+ """The domain this hub's names are for: --names-domain, else its own host."""+ if self._names_domain:+ return self._names_domain+ from urllib.parse import urlsplit+ return urlsplit(self.audience).hostname if self.audience else None++ def names(self) -> "Any":+ """The directory of names this hub serves, read again whenever a file in it changes."""+ from .names import load_directory+ if self.names_dir is None or not self.names_domain:+ return None+ try:+ stamp = tuple(sorted((e.name, e.stat().st_mtime_ns, e.stat().st_size) for e in self.names_dir.iterdir()))+ except OSError:+ return None+ if self._names_cache is None or self._names_cache[0] != stamp:+ self._names_cache = (stamp, load_directory(self.names_dir, self.names_domain))+ return self._names_cache[1]+def _signer(self, method: str, path: str, body: dict[str, Any] | None, headers: dict[str, str]) -> tuple[identity.Signer | None, set[str], tuple[int, Any] | None]: """For a key-signed request: who signed it, what they may do here, or the refusal.""" if not identity.has_key_signature(headers):@@ -670,7 +698,19 @@ class Hub:if method == "GET" and path.startswith("/.well-known/pyrtan/id/") and path.endswith(".json"): ident = identity.PREFIX + path[len("/.well-known/pyrtan/id/"):-len(".json")] doc = self.grants.document(ident) if len(ident) == len(identity.PREFIX) + identity.ID_LENGTH else None+ named = self.names()+ if named is not None:+ held = [d for d in named.docs.values() if d["id"] == ident and (doc is None or d["updated"] > doc["updated"])]+ doc = max(held, key=lambda d: d["updated"]) if held else docreturn (200, doc) if doc else (404, {"status": "error", "reason": "no-identity", "say": "This hub holds no document for that identity."})+ if method == "GET" and (path == "/.well-known/pyrtan/names.json" or (path.startswith("/.well-known/pyrtan/name/") and path.endswith(".json"))):+ named = self.names()+ if named is None:+ return 404, {"status": "error", "reason": "no-names", "say": "This hub serves no names; start it with --names DIR."}+ if path.endswith("/names.json"):+ return 200, {"pyrtan-names": 1, "domain": self.names_domain, "names": named.claims}+ claim = named.claims.get(path[len("/.well-known/pyrtan/name/"):-len(".json")])+ return (200, claim) if claim else (404, {"status": "error", "reason": "no-name", "say": f"No such name at {self.names_domain}."})if method == "GET" and path == "/capabilities": return 200, self.capabilities() if method == "GET" and path == "/gaps":@@ -692,7 +732,7 @@ class Hub:if method == "GET" and path == "/pyrtan.js": return 200, ("text/javascript; charset=utf-8", PYRTAN_JS) if method == "GET" and path == "/api":- endpoints = ["/health", "/peers", "/capabilities", "/intent", "/run", "/resume", "/gaps", "/unanswered", "/watches", "/register", "/allow", "/disallow", "/identity", "/.well-known/pyrtan/id/{id}.json"]+ endpoints = ["/health", "/peers", "/capabilities", "/intent", "/run", "/resume", "/gaps", "/unanswered", "/watches", "/register", "/allow", "/disallow", "/identity", "/.well-known/pyrtan/id/{id}.json", "/.well-known/pyrtan/name/{name}.json", "/.well-known/pyrtan/names.json"]if self.lobby is not None: endpoints += ["/chat", "/chat/start", "/chat/poll", "/chat/say", "/chat/typing", "/chat/settings", "/chat/next", "/chat/leave"] if fed is not None:@@ -789,6 +829,8 @@ def serve_hub(chat: bool = False, allow: list[str] | None = None, grants: str | None = None,+ names: str | Path | None = None,+ names_domain: str | None = None,chat_typing: bool = True, chat_public: bool = False, ) -> tuple[Hub, Server]:@@ -804,7 +846,7 @@ def serve_hub(for who, may in specs: granted.allow(who, may) local = Service(Registry.from_paths(trees), name=f"{name}", env=env, gaps=gaps, egress=egress, watch=watch) if trees else None- hub = Hub(name=name, poll_every=poll_every, token=token, local=local, handout=Registry.from_paths(handout) if handout else None, unanswered=unanswered, chat=chat, chat_typing=chat_typing, chat_public=chat_public, grants=granted)+ hub = Hub(name=name, poll_every=poll_every, token=token, local=local, handout=Registry.from_paths(handout) if handout else None, unanswered=unanswered, chat=chat, chat_typing=chat_typing, chat_public=chat_public, grants=granted, names=names, names_domain=names_domain)server = Server(hub.route, host, port, quiet, token=token, cors=cors, federated=federate or bool(advertise), public=is_chat_path if chat_public else None, keyed=True).start() hub._audience = (advertise or server.url).rstrip("/") if federate or advertise or join: diff --git a/src/pyrtan/identity.py b/src/pyrtan/identity.py index 54f2981..80bb54e 100644 --- a/src/pyrtan/identity.py +++ b/src/pyrtan/identity.py@@ -43,6 +43,10 @@ VERSIONS = {VERSION_ED25519: 32}CHECKSUM_TAG = b".pyrtan id" DOC_TAG = b"pyrtan-doc-v1\n" REQUEST_TAG = "pyrtan-req-v1"+BACKUP_TAG = b"pyrtan-key-backup-v1\n"+HANDOVER_TAG = b"pyrtan-handover-v1\n"+BACKUP_ITERATIONS = 600_000 # PBKDF2-SHA256, as OWASP advises for it in 2023 and after+MIN_ITERATIONS = 100_000 # what a backup may claim, at the least, and still be readID_LENGTH = 56 WINDOW = 300 # seconds a document's time may run ahead of ours YEAR = 365 * 24 * 3600 # the longest a device note may last@@ -477,14 +481,18 @@ def _reissue(doc: dict[str, Any], root: Key, **changes: Any) -> dict[str, Any]:# Name claims.-def make_name(key: Key, name: str, *, issued: int | None = None, **extra: Any) -> dict[str, Any]:- """``key`` claiming ``name`` (``local@domain``). ``extra`` is a domain's own fields (pyrtan.com's ``github``)."""+def make_name(key: Key, name: str, *, issued: int | None = None, handover_from: Key | None = None, **extra: Any) -> dict[str, Any]:+ """``key`` claiming ``name`` (``local@domain``). ``extra`` is a domain's own fields (pyrtan.com's ``github``).+ ``handover_from`` is the key that holds the name now: it signs the claim's body first, as+ ``handover``, which is how a name moves to a new key where names are owned by their key."""parsed = parse(name) if parsed.form != "name": raise IdentityError(f"{name!r} is not a name (name@domain)") claim = {"pyrtan-name": 1, "name": parsed.name, "domain": parsed.domain, "id": key.id, "issued": int(time.time()) if issued is None else issued, **extra} _plain(claim)+ if handover_from is not None:+ claim["handover"] = {"id": handover_from.id, "sig": _b64(handover_from.sign(HANDOVER_TAG + canonical(claim)))}return sign_document(claim, key)@@ -505,6 +513,47 @@ def check_name(claim: Any, *, name: str | None = None) -> str | None:return check_signature(claim, claim["id"])+def check_handover(claim: dict[str, Any], old: str) -> str | None:+ """None if ``claim`` carries ``old``'s handover; otherwise why not."""+ h = claim.get("handover")+ if not isinstance(h, dict) or not isinstance(h.get("sig"), str):+ return f"the name moves from {short(old)} to {short(claim.get('id', '?'))} without a handover signed by {short(old)}"+ if h.get("id") != old:+ return f"the handover is {short(str(h.get('id')))}'s, not {short(old)}'s"+ body = {k: v for k, v in claim.items() if k not in ("sig", "handover")}+ try:+ ok = verify(old, HANDOVER_TAG + canonical(body), _unb64(h["sig"]))+ except (IdentityError, ValueError):+ ok = False+ return None if ok else f"the handover is not signed by {short(old)}"+++def make_release(key: Key, name: str, *, issued: int | None = None) -> dict[str, Any]:+ """``key`` giving up ``name``: what replaces its claim, so the name is free again."""+ parsed = parse(name)+ if parsed.form != "name":+ raise IdentityError(f"{name!r} is not a name (name@domain)")+ release = {"pyrtan-name-release": 1, "name": parsed.name, "domain": parsed.domain, "id": key.id,+ "issued": int(time.time()) if issued is None else issued}+ return sign_document(release, key)+++def check_release(release: Any, *, name: str | None = None) -> str | None:+ if not isinstance(release, dict) or release.get("pyrtan-name-release") != 1:+ return "not a name release (pyrtan-name-release: 1)"+ try:+ normal(release.get("id"))+ _int(release, "issued")+ stated = parse(f"{release.get('name')}@{release.get('domain')}")+ except IdentityError as e:+ return str(e)+ if stated.form != "name" or (stated.name, stated.domain) != (release["name"], release["domain"]):+ return "name and domain are not a lowercase name@domain"+ if name is not None and str(parse(name)) != str(stated):+ return f"the release is for {stated}, not {name}"+ return check_signature(release, release["id"])++def check_any(doc: Any, *, now: float | None = None) -> tuple[str, str | None]: """Which kind of document this is, and None or what is wrong with it.""" if isinstance(doc, dict):@@ -514,7 +563,11 @@ def check_any(doc: Any, *, now: float | None = None) -> tuple[str, str | None]:return "device", check_device(doc, now=now) if "pyrtan-name" in doc: return "name", check_name(doc)- return "unknown", "not a pyrtan identity document, device note or name claim"+ if "pyrtan-name-release" in doc:+ return "release", check_release(doc)+ if "pyrtan-key-backup" in doc:+ return "backup", backup_problem(doc)+ return "unknown", "not a pyrtan identity document, device note, name claim, release or key backup"# ---------------------------------------------------------------- signed requests@@ -594,6 +647,93 @@ def verify_request(audience: str, method: str, path: str, body: Any, headers: direturn Signer(key, note["root"], note)+# ---------------------------------------------------------------- backups+#+# A root key, encrypted with a passphrase, beside the identity document it+# signs, so that a person can keep it somewhere durable (a file, their own+# private repository) and restore it on another machine or browser. The+# console writes the same format with WebCrypto (identity.js). It is for+# places only its owner can read: an encrypted key where anyone can copy+# it can be guessed at offline, forever.+++def _aesgcm_kdf():+ try:+ from cryptography.hazmat.primitives import hashes+ from cryptography.hazmat.primitives.ciphers.aead import AESGCM+ from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2HMAC+ except ImportError:+ raise IdentityError(INSTALL) from None+ return hashes, AESGCM, PBKDF2HMAC+++def _backup_key(passphrase: str, salt: bytes, iterations: int) -> bytes:+ hashes, _, PBKDF2HMAC = _aesgcm_kdf()+ return PBKDF2HMAC(algorithm=hashes.SHA256(), length=32, salt=salt, iterations=iterations).derive(passphrase.encode("utf-8"))+++def make_backup(key: Key, passphrase: str, *, document: dict[str, Any] | None = None, salt: bytes | None = None,+ iv: bytes | None = None, iterations: int = BACKUP_ITERATIONS, made: int | None = None) -> dict[str, Any]:+ """``key`` encrypted with ``passphrase`` (PBKDF2-SHA256, then AES-256-GCM over the 32-byte+ seed, the identifier as associated data), with its identity document beside it."""+ if len(passphrase) < 8:+ raise IdentityError("a passphrase is at least 8 characters; a long sentence is better")+ _, AESGCM, _ = _aesgcm_kdf()+ salt = os.urandom(16) if salt is None else salt+ iv = os.urandom(12) if iv is None else iv+ data = AESGCM(_backup_key(passphrase, salt, iterations)).encrypt(iv, key._seed, BACKUP_TAG + key.id.encode("ascii"))+ backup = {+ "pyrtan-key-backup": 1, "id": key.id, "made": int(time.time()) if made is None else made,+ "kdf": {"name": "PBKDF2-SHA256", "iterations": iterations, "salt": _b64(salt)},+ "cipher": {"name": "AES-256-GCM", "iv": _b64(iv)}, "data": _b64(data),+ }+ if document is not None:+ if document.get("id") != key.id:+ raise IdentityError("the document is another identity's")+ backup["document"] = document+ return backup+++def backup_problem(backup: Any) -> str | None:+ """What is wrong with a backup's shape, before any passphrase is tried; None if nothing."""+ if not isinstance(backup, dict) or backup.get("pyrtan-key-backup") != 1:+ return "not a key backup (pyrtan-key-backup: 1)"+ kdf, cipher = backup.get("kdf"), backup.get("cipher")+ try:+ normal(backup.get("id"))+ except IdentityError as e:+ return str(e)+ if not isinstance(kdf, dict) or kdf.get("name") != "PBKDF2-SHA256" or not isinstance(kdf.get("iterations"), int) or not isinstance(kdf.get("salt"), str):+ return "the backup's kdf is not PBKDF2-SHA256 with iterations and a salt"+ if not MIN_ITERATIONS <= kdf["iterations"] <= 10_000_000:+ return f"the backup's kdf claims {kdf['iterations']} iterations; at least {MIN_ITERATIONS} are read"+ if not isinstance(cipher, dict) or cipher.get("name") != "AES-256-GCM" or not isinstance(cipher.get("iv"), str) or not isinstance(backup.get("data"), str):+ return "the backup's cipher is not AES-256-GCM with an iv and data"+ doc = backup.get("document")+ if doc is not None:+ problem = check_identity(doc, ident=backup["id"])+ if problem:+ return f"the document beside the key: {problem}"+ return None+++def open_backup(backup: Any, passphrase: str) -> Key:+ """The key in ``backup``; IdentityError for a wrong passphrase or a damaged file."""+ problem = backup_problem(backup)+ if problem:+ raise IdentityError(problem)+ _, AESGCM, _ = _aesgcm_kdf()+ try:+ salt, iv, data = _unb64(backup["kdf"]["salt"]), _unb64(backup["cipher"]["iv"]), _unb64(backup["data"])+ seed = AESGCM(_backup_key(passphrase, salt, backup["kdf"]["iterations"])).decrypt(iv, data, BACKUP_TAG + backup["id"].lower().encode("ascii"))+ except Exception:+ raise IdentityError("the passphrase does not open this backup (or the file is damaged)") from None+ key = Key(seed)+ if key.id != backup["id"].lower():+ raise IdentityError("the backup's key is not the identity it names")+ return key++# ---------------------------------------------------------------- where the CLI keeps them diff --git a/src/pyrtan/names.py b/src/pyrtan/names.py index 66fff3d..253ce52 100644 --- a/src/pyrtan/names.py +++ b/src/pyrtan/names.py@@ -8,13 +8,20 @@ the first time. A different answer later is never followed; it is shownbeside the pinned one and the person re-pins by hand, as SSH does with a changed host key. That is what makes a domain's word safe to take.-Any domain can be a name provider by serving those files.+Any domain can be a name provider by serving those files, from any+static host or from its own hubs (``pyrtan hub --names DIR``).:func:`build_wellknown` writes them from a directory of claims (and of identity documents, served at ``/.well-known/pyrtan/id/<56>.json``);-pyrtan.com's build runs it on ``site/names/``. :func:`gate` is pyrtan.com's-own rule for that directory: a name is its holder's GitHub login, added-and changed only by the GitHub account whose numeric id the claim-carries, because GitHub frees a login when its owner renames.+pyrtan.com's build runs it on ``site/names/``.++Who may change a name is a directory's own rule. :func:`gate_key`, for any+network on any git host, needs nothing but signatures: a new name is open+to whoever signs it, a name moves to a new key only with a *handover* its+old key signs, it is given up only by a *release* its key signs, and an+identity document is never replaced by an older one or deleted while its+name stands. :func:`gate` is pyrtan.com's own rule: a name is its holder's+GitHub login, added and changed only by the GitHub account whose numeric+id the claim carries, because GitHub frees a login when its owner renames.This module never fetches an ``.onion``: that needs Tor, which pyrtan does not reach yet. The design is ``_meta/identity.md`` §6.@@ -26,6 +33,7 @@ import jsonimport re import shutil import time+from dataclasses import dataclass, fieldfrom pathlib import Path from typing import Any, Callable@@ -140,50 +148,106 @@ class Pins:# ---------------------------------------------------------------- a directory of names-def read_directory(directory: Path, domain: str, *, label: str | None = None) -> tuple[dict[str, dict[str, Any]], dict[str, dict[str, Any]], list[Problem]]:- """The claims (``<name>.json``) and identity documents (``<name>.id.json``) in- ``directory``, for ``domain``: ``(claims by name, documents by name, problems)``.- Problems name files as ``label/<file>`` (default: the directory as given)."""- claims: dict[str, dict[str, Any]] = {}- docs: dict[str, dict[str, Any]] = {}- problems: list[Problem] = []- if not directory.is_dir():- return claims, docs, problems+@dataclass+class Directory:+ """What a directory of names holds: claims, identity documents and releases, by name."""+ claims: dict[str, dict[str, Any]] = field(default_factory=dict)+ docs: dict[str, dict[str, Any]] = field(default_factory=dict)+ releases: dict[str, dict[str, Any]] = field(default_factory=dict)+ problems: list[Problem] = field(default_factory=list)+++def load_directory(directory: Path | None, domain: str, *, label: str | None = None) -> Directory:+ """The claims (``<name>.json``), releases (a ``<name>.json`` that gives the name up) and+ identity documents (``<name>.id.json``) in ``directory``, for ``domain``. Problems name+ files as ``label/<file>`` (default: the directory as given); what fails a check is left out."""+ d = Directory()+ if directory is None or not directory.is_dir():+ return dlabel = directory.as_posix() if label is None else label def rel(p: Path) -> str: return f"{label}/{p.name}"+ raw_docs: dict[str, Any] = {}for path in sorted(directory.iterdir()): if path.name.startswith(".") or path.name == "README.md": continue if not path.is_file() or not path.name.endswith(".json"):- problems.append(Problem(rel(path), "only <name>.json and <name>.id.json files belong here"))+ d.problems.append(Problem(rel(path), "only <name>.json and <name>.id.json files belong here"))continue try: data = json.loads(path.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError) as e:- problems.append(Problem(rel(path), f"not JSON: {e}"))+ d.problems.append(Problem(rel(path), f"not JSON: {e}"))continue if path.name.endswith(".id.json"):- docs[path.name[: -len(".id.json")]] = data+ raw_docs[path.name[: -len(".id.json")]] = datacontinue stem = path.name[: -len(".json")]- problem = identity.check_name(data, name=f"{stem}@{domain}") if identity.LOCAL.match(stem) else f"{stem!r} is not a name"+ if not identity.LOCAL.match(stem):+ d.problems.append(Problem(rel(path), f"{stem!r} is not a name"))+ continue+ if isinstance(data, dict) and "pyrtan-name-release" in data:+ problem = identity.check_release(data, name=f"{stem}@{domain}")+ target = d.releases+ else:+ problem = identity.check_name(data, name=f"{stem}@{domain}")+ target = d.claimsif problem:- problems.append(Problem(rel(path), problem))+ d.problems.append(Problem(rel(path), problem))continue- claims[stem] = data- for stem, doc in docs.items():+ target[stem] = data+ for stem, doc in raw_docs.items():where = rel(directory / f"{stem}.id.json")- if stem not in claims:- problems.append(Problem(where, f"an identity document needs its name's claim, {stem}.json, beside it"))+ if stem not in d.claims:+ d.problems.append(Problem(where, f"an identity document needs its name's claim, {stem}.json, beside it"))continue- problem = identity.check_identity(doc, ident=claims[stem]["id"])+ problem = identity.check_identity(doc, ident=d.claims[stem]["id"])if problem:- problems.append(Problem(where, problem))- docs = {s: d for s, d in docs.items() if s in claims and identity.check_identity(d, ident=claims[s]["id"]) is None}- return claims, docs, problems+ d.problems.append(Problem(where, problem))+ continue+ d.docs[stem] = doc+ return d+++def read_directory(directory: Path, domain: str, *, label: str | None = None) -> tuple[dict[str, dict[str, Any]], dict[str, dict[str, Any]], list[Problem]]:+ """``(claims by name, documents by name, problems)``; see :func:`load_directory`."""+ d = load_directory(directory, domain, label=label)+ return d.claims, d.docs, d.problems+++def gate_key(head: Path, base: Path | None, domain: str, *, label: str = "names") -> list[Problem]:+ """The rule for a directory of names owned by their keys, on any git host: ``head`` is the+ directory a change proposes, ``base`` the one it changes (None for a first commit).+ Needs only the two trees, never who proposed the change."""+ h = load_directory(head, domain, label=label)+ b = load_directory(base, domain, label=label)+ problems = list(h.problems)+ names = sorted(set(h.claims) | set(h.releases) | set(b.claims) | set(b.releases) | set(b.docs))+ for n in names:+ where, dwhere = f"{label}/{n}.json", f"{label}/{n}.id.json"+ held, hc, hr = b.claims.get(n), h.claims.get(n), h.releases.get(n)+ if held is not None:+ if hc is not None and hc["id"] != held["id"]:+ problem = identity.check_handover(hc, held["id"])+ if problem:+ problems.append(Problem(where, problem))+ elif hr is not None and hr["id"] != held["id"]:+ problems.append(Problem(where, f"the release is {identity.short(hr['id'])}'s; {n} is held by {identity.short(held['id'])}"))+ elif hc is None and hr is None and not (head / f"{n}.json").exists():+ problems.append(Problem(where, f"{n} is given up by replacing its claim with a release its key signs (pyrtan id release), not by deleting it"))+ elif hr is not None and n not in b.releases:+ problems.append(Problem(where, f"nobody holds {n}; there is nothing to release"))+ # An identity document only moves forward while its name stays with the same key.+ old = b.docs.get(n)+ if old is not None and held is not None and hc is not None and hc["id"] == held["id"]:+ new = h.docs.get(n)+ if new is None and not (head / f"{n}.id.json").exists():+ problems.append(Problem(dwhere, "an identity document is replaced by a newer one its key signs, not deleted: deleting it would hide its revocations"))+ elif new is not None and new["updated"] < old["updated"]:+ problems.append(Problem(dwhere, "this document is older than the one it replaces: an older one could bring a revoked device back"))+ return problemsdef build_wellknown(directory: Path, out: Path, domain: str) -> tuple[int, list[Problem]]:@@ -276,11 +340,16 @@ def _github_of(raw: bytes | None) -> int | None:def place(checkout: Path, claim: dict[str, Any], doc: dict[str, Any] | None) -> list[Path]:- """Write a claim (and the identity document) into ``checkout/site/names/``."""- directory = checkout / NAMES_DIR- if not (checkout / "site").is_dir():- raise IdentityError(f"{checkout}: no site/ directory; is this a checkout of github.com/b-i-t-8/pyrtan?")- domain = (checkout / "site" / "CNAME").read_text(encoding="utf-8").strip() if (checkout / "site" / "CNAME").is_file() else "pyrtan.com"+ """Write a claim (and the identity document) where ``checkout`` keeps names: a network's+ names repository (``pyrtan-names.json`` and ``names/``), or this repository's ``site/names/``."""+ network = repo_domain(checkout)+ if network is not None:+ directory, domain = checkout / "names", network+ elif (checkout / "site").is_dir():+ directory = checkout / NAMES_DIR+ domain = (checkout / "site" / "CNAME").read_text(encoding="utf-8").strip() if (checkout / "site" / "CNAME").is_file() else "pyrtan.com"+ else:+ raise IdentityError(f"{checkout}: neither a names repository (pyrtan names init) nor a checkout of github.com/b-i-t-8/pyrtan")if claim["domain"] != domain: raise IdentityError(f"this checkout serves names at {domain}; the claim is for {claim['name']}@{claim['domain']}") directory.mkdir(parents=True, exist_ok=True)@@ -302,3 +371,211 @@ def github_id(login: str, *, fetch: Fetch | None = None) -> int:if status != 200 or not isinstance(body, dict) or not isinstance(body.get("id"), int): raise IdentityError(f"{url} answered {status}; pass --github YOUR-NUMERIC-ID") return body["id"]+++# ---------------------------------------------------------------- a network's own names repository++REPO_FILE = "pyrtan-names.json" # at a names repository's root: {"pyrtan-names-repo": 1, "domain": ...}+HOSTS = ("github", "gitlab", "plain")+PYRTAN_GIT = "https://github.com/b-i-t-8/pyrtan"++_README = """# Names at __DOMAIN__++A name here is `<name>@__DOMAIN__`, pointing at a pyrtan identity (a key:+`pyr:` and 56 characters). Anyone resolves it with++ pyrtan id resolve <name>@__DOMAIN__++which fetches `https://__DOMAIN__/.well-known/pyrtan/name/<name>.json`, checks+that the key signed it, and pins the answer. The names are the files in+`names/`; they are served by GitHub or GitLab Pages, or by any hub of the+network started `pyrtan hub --names names --names-domain __DOMAIN__`.++## Getting one++In the pyrtan console, Identity tab: *Claim a name*, with this repository and+`__DOMAIN__`. Or from a shell:++ pyrtan id name <name>@__DOMAIN__ --into <a checkout of this repository>++Then open a pull (or merge) request with `names/<name>.json`, and+`names/<name>.id.json` if you want hubs to find your devices here.++## Who may change a name++Names here are owned by their keys, and the gate checks only signatures:++- a new name is open to whoever signs it, while nobody holds it;+- a name moves to a new key only with a handover its old key signs+ (`pyrtan id name <name>@__DOMAIN__ --handover-from OLD-KEY`);+- a name is given up by replacing its claim with a release its key signs+ (`pyrtan id release <name>@__DOMAIN__`), never by deleting it;+- an identity document is replaced only by a newer one its key signs.++Nothing in this repository is secret: never put an invite, a token or anyone's+key backup here, not even encrypted.+"""++_GITHUB_GATE = """# Generated by `pyrtan names init`: the names gate for __DOMAIN__.+#+# pull_request_target, so the gate is always this file as it is on the default+# branch, and pyrtan comes from a pinned source, never from the pull request: a+# pull request cannot change what judges it. The pull request's tree is checked+# out as data and compared with its merge base; nothing in it is run. Read-only+# token, no secrets, no credentials left in the checkouts, no cache.+# Pin __REF__ to a commit of pyrtan you have read.++name: names-gate++on:+ pull_request_target:+ paths: ["names/**"]++permissions:+ contents: read++jobs:+ gate:+ runs-on: ubuntu-latest+ env:+ BASE_SHA: ${{ github.event.pull_request.base.sha }}+ steps:+ - uses: actions/checkout@v4+ with:+ ref: ${{ github.event.pull_request.base.sha }}+ path: base-repo+ fetch-depth: 0+ persist-credentials: false+ - uses: actions/checkout@v4+ with:+ repository: ${{ github.event.pull_request.head.repo.full_name }}+ ref: ${{ github.event.pull_request.head.sha }}+ path: pr+ fetch-depth: 0+ persist-credentials: false+ - name: The merge base, as data+ run: |+ git -C pr fetch --no-tags "$GITHUB_WORKSPACE/base-repo" "$BASE_SHA"+ git -C pr worktree add --detach "$GITHUB_WORKSPACE/base" "$(git -C pr merge-base "$BASE_SHA" HEAD)"+ - uses: actions/setup-python@v5+ with:+ python-version: "3.12"+ - run: pip install "pyrtan[identity] @ git+__PYRTAN__@__REF__"+ - run: pyrtan names gate pr/names --base base/names --domain __DOMAIN__+"""++_GITHUB_PAGES = """# Generated by `pyrtan names init`: serve names/ as https://__DOMAIN__/.well-known/pyrtan/.+# Set the repository's Pages source to "GitHub Actions" and its custom domain to __DOMAIN__.++name: names-pages++on:+ push:+ branches: [main]+ workflow_dispatch:++permissions:+ contents: read+ pages: write+ id-token: write++concurrency:+ group: pages+ cancel-in-progress: true++jobs:+ build:+ runs-on: ubuntu-latest+ steps:+ - uses: actions/checkout@v4+ - uses: actions/setup-python@v5+ with:+ python-version: "3.12"+ - run: pip install "pyrtan[identity] @ git+__PYRTAN__@__REF__"+ - run: pyrtan id wellknown names --domain __DOMAIN__ --out site && echo __DOMAIN__ > site/CNAME+ # v3, not v4: v4 leaves out dot-directories, and the names live in .well-known/.+ - uses: actions/upload-pages-artifact@v3+ with:+ path: site++ deploy:+ needs: build+ runs-on: ubuntu-latest+ environment:+ name: github-pages+ url: ${{ steps.deployment.outputs.page_url }}+ steps:+ - id: deployment+ uses: actions/deploy-pages@v4+"""++_GITLAB = """# Generated by `pyrtan names init`: the names gate for __DOMAIN__, and Pages.+#+# On GitLab a merge request's pipeline runs the merge request's own copy of this+# file, so read any change to it before merging: the gate is pyrtan from a pinned+# source, not from the merge request, but this file says which. Protect the+# default branch and require the pipeline to pass. Pin __REF__ to a commit of+# pyrtan you have read.++variables:+ GIT_DEPTH: "0"++names-gate:+ image: python:3.12+ rules:+ - if: $CI_PIPELINE_SOURCE == "merge_request_event"+ changes: ["names/**/*"]+ script:+ - pip install "pyrtan[identity] @ git+__PYRTAN__@__REF__"+ - git fetch origin "$CI_MERGE_REQUEST_TARGET_BRANCH_NAME"+ - git worktree add --detach /tmp/base "$(git merge-base FETCH_HEAD HEAD)"+ - pyrtan names gate names --base /tmp/base/names --domain __DOMAIN__++pages:+ image: python:3.12+ rules:+ - if: $CI_COMMIT_BRANCH == $CI_DEFAULT_BRANCH+ script:+ - pip install "pyrtan[identity] @ git+__PYRTAN__@__REF__"+ - pyrtan id wellknown names --domain __DOMAIN__ --out public+ artifacts:+ paths: [public]+"""+++def init_repo(directory: Path, domain: str, *, host: str = "github", pyrtan_ref: str = "main") -> tuple[list[Path], list[Path]]:+ """Write a names repository for ``domain`` into ``directory``: ``pyrtan-names.json``, a+ README, ``names/``, and the gate and Pages for ``host``. Returns ``(written, kept)``: a file+ that is already there is kept, never overwritten."""+ domain = identity.check_host(domain.lower())+ if host not in HOSTS:+ raise IdentityError(f"host is one of {', '.join(HOSTS)}")+ if not re.fullmatch(r"[A-Za-z0-9._/-]{1,80}", pyrtan_ref):+ raise IdentityError(f"{pyrtan_ref!r} is not a git ref")+ fill = lambda t: t.replace("__DOMAIN__", domain).replace("__REF__", pyrtan_ref).replace("__PYRTAN__", PYRTAN_GIT) # noqa: E731+ files = {REPO_FILE: json.dumps({"pyrtan-names-repo": 1, "domain": domain}, indent=2) + "\n", "README.md": fill(_README), "names/.gitkeep": ""}+ if host == "github":+ files[".github/workflows/names-gate.yml"] = fill(_GITHUB_GATE)+ files[".github/workflows/names-pages.yml"] = fill(_GITHUB_PAGES)+ elif host == "gitlab":+ files[".gitlab-ci.yml"] = fill(_GITLAB)+ written, kept = [], []+ for rel, text in files.items():+ path = directory / rel+ if path.exists():+ kept.append(path)+ continue+ path.parent.mkdir(parents=True, exist_ok=True)+ path.write_text(text, encoding="utf-8")+ written.append(path)+ return written, kept+++def repo_domain(checkout: Path) -> str | None:+ """The domain a names repository (``pyrtan-names.json``) is for, if ``checkout`` is one."""+ try:+ data = json.loads((checkout / REPO_FILE).read_text(encoding="utf-8"))+ except (OSError, json.JSONDecodeError):+ return None+ return data.get("domain") if isinstance(data, dict) and data.get("pyrtan-names-repo") == 1 and isinstance(data.get("domain"), str) else None+diff --git a/tests/test_identity.py b/tests/test_identity.py index 09e9469..2b93c5f 100644 --- a/tests/test_identity.py +++ b/tests/test_identity.py@@ -316,3 +316,66 @@ def test_cli_check_identifiers(tmp_path, capsys):def test_cli_without_an_identity(tmp_path, capsys): code, _, err = run(capsys, "id", "show", "--dir", str(tmp_path / "nobody")) assert code == 1 and "pyrtan id new" in err+++# ---------------------------------------------------------------- backups, releases, handovers+++@needs_crypto+def test_backups(root, device):+ doc = identity.make_identity(root, updated=int(time.time()) - 10)+ b = identity.make_backup(root, "correct horse battery", document=doc, iterations=identity.MIN_ITERATIONS)+ assert identity.backup_problem(b) is None and identity.check_any(b) == ("backup", None)+ assert identity.open_backup(b, "correct horse battery").id == root.id+ with pytest.raises(IdentityError, match="passphrase does not open"):+ identity.open_backup(b, "wrong horse battery")+ # The ciphertext is bound to the identity it names, and a weakened kdf is refused.+ with pytest.raises(IdentityError, match="passphrase does not open"):+ identity.open_backup({**b, "id": device.id, "document": None}, "correct horse battery")+ assert "iterations" in identity.backup_problem({**b, "kdf": {**b["kdf"], "iterations": 1000}})+ assert "document" in identity.backup_problem({**b, "document": identity.make_identity(device)})+ with pytest.raises(IdentityError, match="at least 8"):+ identity.make_backup(root, "short")+ assert identity.make_backup(root, "correct horse battery", iterations=identity.MIN_ITERATIONS)["kdf"]["salt"] != b["kdf"]["salt"]+++@needs_crypto+def test_releases_and_handovers(root, device):+ release = identity.make_release(root, "alice@example.org", issued=NOW)+ assert identity.check_release(release, name="alice@example.org") is None+ assert "bob" in identity.check_release(release, name="bob@example.org")+ assert "signature" in identity.check_release({**release, "id": device.id})+ claim = identity.make_name(device, "alice@example.org", issued=NOW)+ assert "without a handover" in identity.check_handover(claim, root.id)+ moved = identity.make_name(device, "alice@example.org", issued=NOW, handover_from=root)+ assert identity.check_name(moved) is None # still the new key's own claim+ assert identity.check_handover(moved, root.id) is None+ assert "not" in identity.check_handover(moved, device.id)+ assert "not signed" in identity.check_handover({**moved, "issued": NOW + 1}, root.id)+++@needs_crypto+def test_cli_backup_and_restore(tmp_path, capsys, monkeypatch):+ monkeypatch.setenv("PP", "correct horse battery staple")+ a, c = str(tmp_path / "a"), str(tmp_path / "c")+ run(capsys, "id", "new", "--dir", a)+ phone = identity.Key.generate().id+ run(capsys, "id", "device", "add", phone, "--dir", a)+ code, out, _ = run(capsys, "id", "backup", "--out", str(tmp_path / "b.json"), "--passphrase-env", "PP", "--dir", a)+ assert code == 0 and "never in a public one" in out+ assert stat.S_IMODE(os.stat(tmp_path / "b.json").st_mode) == 0o600+ assert run(capsys, "id", "backup", "--out", str(tmp_path / "b.json"), "--passphrase-env", "PP", "--dir", a)[0] == 1 # not overwritten+ code, out, _ = run(capsys, "id", "check", str(tmp_path / "b.json"))+ assert code == 0 and "a key backup of" in out+ code, out, _ = run(capsys, "id", "restore", str(tmp_path / "b.json"), "--passphrase-env", "PP", "--dir", c)+ assert code == 0 and "again" in out+ code, out, _ = run(capsys, "id", "show", "--dir", c)+ assert phone in out # the document came back with the key+ monkeypatch.setenv("PP", "not the passphrase")+ code, _, err = run(capsys, "id", "restore", str(tmp_path / "b.json"), "--passphrase-env", "PP", "--dir", str(tmp_path / "d"))+ assert code == 1 and "does not open" in err+ other = str(tmp_path / "other")+ run(capsys, "id", "new", "--dir", other)+ monkeypatch.setenv("PP", "correct horse battery staple")+ code, _, err = run(capsys, "id", "restore", str(tmp_path / "b.json"), "--passphrase-env", "PP", "--dir", other)+ assert code == 1 and "already holds" in errdiff --git a/tests/test_names.py b/tests/test_names.py index 055d257..4ab9726 100644 --- a/tests/test_names.py +++ b/tests/test_names.py@@ -1,4 +1,5 @@import json+import shutilimport subprocess import sys@@ -269,3 +270,165 @@ def test_cli_wellknown(tmp_path, capsys, alice):(d / "bad.json").write_text("{}") code, _, err = run(capsys, "id", "wellknown", str(d), "--domain", "example.org", "--out", str(tmp_path / "site")) assert code == 1 and "bad.json" in err+++# ---------------------------------------------------------------- a network's names, owned by their keys++DOMAIN = "names.example.org"+++def write(d, name, doc):+ d.mkdir(parents=True, exist_ok=True)+ (d / name).write_text(json.dumps(doc))+++def test_key_rule_new_names_and_handovers(tmp_path, alice, bob):+ base, head = tmp_path / "base", tmp_path / "head"+ write(head, "alice.json", identity.make_name(alice, f"alice@{DOMAIN}"))+ assert names.gate_key(head, None, DOMAIN) == [] # a new name: anyone who signs it+ shutil.copytree(head, base)+ write(head, "alice.json", identity.make_name(alice, f"alice@{DOMAIN}"))+ assert names.gate_key(head, base, DOMAIN) == [] # re-signed by the same key+ write(head, "alice.json", identity.make_name(bob, f"alice@{DOMAIN}"))+ assert "without a handover" in messages(names.gate_key(head, base, DOMAIN))+ write(head, "alice.json", identity.make_name(bob, f"alice@{DOMAIN}", handover_from=bob))+ assert "not" in messages(names.gate_key(head, base, DOMAIN))+ write(head, "alice.json", identity.make_name(bob, f"alice@{DOMAIN}", handover_from=alice))+ assert names.gate_key(head, base, DOMAIN) == [] # moved, with alice's handover+ write(head, "alice.json", identity.make_name(alice, "alice@elsewhere.org"))+ assert "the claim is for alice@elsewhere.org" in messages(names.gate_key(head, base, DOMAIN))+++def test_key_rule_releases_and_deletions(tmp_path, alice, bob):+ base, head = tmp_path / "base", tmp_path / "head"+ write(base, "alice.json", identity.make_name(alice, f"alice@{DOMAIN}"))+ head.mkdir()+ assert "not by deleting it" in messages(names.gate_key(head, base, DOMAIN))+ write(head, "alice.json", identity.make_release(bob, f"alice@{DOMAIN}"))+ assert "is held by" in messages(names.gate_key(head, base, DOMAIN))+ write(head, "alice.json", identity.make_release(alice, f"alice@{DOMAIN}"))+ assert names.gate_key(head, base, DOMAIN) == [] # given up by its key+ # Released, the name is free: anyone may claim it.+ shutil.rmtree(base)+ shutil.copytree(head, base)+ write(head, "alice.json", identity.make_name(bob, f"alice@{DOMAIN}"))+ assert names.gate_key(head, base, DOMAIN) == []+ write(head, "carol.json", identity.make_release(bob, f"carol@{DOMAIN}"))+ assert "nothing to release" in messages(names.gate_key(head, base, DOMAIN))+ # A release is not served as a name.+ count, _ = names.build_wellknown(base, tmp_path / "out", DOMAIN)+ assert count == 0+++def test_key_rule_documents_only_move_forward(tmp_path, alice):+ base, head = tmp_path / "base", tmp_path / "head"+ claim = identity.make_name(alice, f"alice@{DOMAIN}")+ old = identity.make_identity(alice, updated=1_700_000_000)+ new = identity.make_identity(alice, updated=1_700_000_100, revoked=[identity.Key.generate().id])+ write(base, "alice.json", claim)+ write(base, "alice.id.json", new)+ write(head, "alice.json", claim)+ write(head, "alice.id.json", old)+ assert "older than the one it replaces" in messages(names.gate_key(head, base, DOMAIN))+ (head / "alice.id.json").unlink()+ assert "not deleted" in messages(names.gate_key(head, base, DOMAIN))+ write(head, "alice.id.json", identity.make_identity(alice, updated=1_700_000_200))+ assert names.gate_key(head, base, DOMAIN) == []+++def test_names_init_and_the_github_gate_steps(tmp_path, alice, bob):+ """names init writes a repository; the gate workflow's merge-base step, against real git,+ judges only what a pull request changed, not what landed on main after it branched."""+ repo = tmp_path / "repo"+ written, kept = names.init_repo(repo, DOMAIN, host="github", pyrtan_ref="0123abc")+ rels = sorted(p.relative_to(repo).as_posix() for p in written)+ assert rels == [".github/workflows/names-gate.yml", ".github/workflows/names-pages.yml", "README.md", "names/.gitkeep", "pyrtan-names.json"]+ gate_yml = (repo / ".github/workflows/names-gate.yml").read_text()+ assert "pull_request_target" in gate_yml and "@0123abc" in gate_yml and DOMAIN in gate_yml and "merge-base" in gate_yml+ assert "upload-pages-artifact@v3" in (repo / ".github/workflows/names-pages.yml").read_text()+ assert names.repo_domain(repo) == DOMAIN+ assert names.init_repo(repo, DOMAIN)[1] # nothing overwritten the second time+ with pytest.raises(IdentityError):+ names.init_repo(tmp_path / "x", DOMAIN, pyrtan_ref="main; rm -rf /")+ git = lambda *a: subprocess.run(["git", "-c", "user.name=t", "-c", "user.email=t@t", *a], cwd=repo, check=True, capture_output=True, text=True)+ git("init", "-q", "-b", "main")+ write(repo / "names", "alice.json", identity.make_name(alice, f"alice@{DOMAIN}"))+ git("add", ".")+ git("commit", "-q", "-m", "alice")+ git("checkout", "-q", "-b", "pr")+ write(repo / "names", "bob.json", identity.make_name(bob, f"bob@{DOMAIN}"))+ git("add", ".")+ git("commit", "-q", "-m", "bob")+ git("checkout", "-q", "main")+ write(repo / "names", "carol.json", identity.make_name(identity.Key.generate(), f"carol@{DOMAIN}"))+ git("add", ".")+ git("commit", "-q", "-m", "carol, after the pull request branched")+ main_sha = git("rev-parse", "HEAD").stdout.strip()+ git("checkout", "-q", "pr")+ base = tmp_path / "base"+ git("worktree", "add", "--detach", str(base), git("merge-base", main_sha, "HEAD").stdout.strip())+ assert names.gate_key(repo / "names", base / "names", DOMAIN) == [] # carol is not a deletion by the pull request+ gate = subprocess.run([sys.executable, "-m", "pyrtan.cli", "names", "gate", str(repo / "names"), "--base", str(base / "names"), "--domain", DOMAIN], capture_output=True, text=True)+ assert gate.returncode == 0 and "open" in gate.stdout+++def test_names_init_for_gitlab(tmp_path):+ written, _ = names.init_repo(tmp_path / "gl", DOMAIN, host="gitlab")+ ci = (tmp_path / "gl" / ".gitlab-ci.yml").read_text()+ assert "merge_request_event" in ci and "merge-base" in ci and "public" in ci+ assert not (tmp_path / "gl" / ".github").exists()+++def test_a_hub_serves_its_networks_names(tmp_path, alice, bob):+ from pyrtan.hub import serve_hub+ from pyrtan.service import http_json+ d = tmp_path / "names"+ claim = identity.make_name(alice, f"alice@{DOMAIN}")+ doc = identity.make_identity(alice, routes=["hub.example"])+ write(d, "alice.json", claim)+ write(d, "alice.id.json", doc)+ write(d, "bob.json", identity.make_release(bob, f"bob@{DOMAIN}"))+ h, srv = serve_hub(port=0, quiet=True, token="tok", names=str(d), names_domain=DOMAIN, poll_every=60)+ try:+ url = srv.url.rstrip("/")+ assert http_json("GET", f"{url}/.well-known/pyrtan/name/alice.json") == (200, claim) # public, behind a token+ assert http_json("GET", f"{url}/.well-known/pyrtan/name/bob.json")[0] == 404 # released+ assert http_json("GET", f"{url}/.well-known/pyrtan/id/{alice.id[4:]}.json") == (200, doc)+ assert list(http_json("GET", f"{url}/.well-known/pyrtan/names.json")[1]["names"]) == ["alice"]+ # A new claim lands (a git pull, say) and is served at once.+ claim_b = identity.make_name(bob, f"bob@{DOMAIN}")+ write(d, "bob.json", claim_b)+ assert http_json("GET", f"{url}/.well-known/pyrtan/name/bob.json") == (200, claim_b)+ # And resolving through it is the same as through any domain.+ fetch = lambda u: http_json("GET", u.replace(f"https://{DOMAIN}", url))+ assert names.resolve(f"alice@{DOMAIN}", fetch=fetch)["id"] == alice.id+ finally:+ h.stop()+ srv.stop()+++def test_cli_names_and_releases(tmp_path, capsys, monkeypatch):+ repo = tmp_path / "repo"+ code, out, _ = run(capsys, "names", "init", str(repo), "--domain", DOMAIN, "--host", "plain")+ assert code == 0 and "pyrtan-names.json" in out and not (repo / ".github").exists()+ me, new = str(tmp_path / "me"), str(tmp_path / "new")+ run(capsys, "id", "new", "--dir", me)+ code, out, err = run(capsys, "id", "name", f"alice@{DOMAIN}", "--into", str(repo), "--dir", me)+ assert code == 0, err+ assert (repo / "names" / "alice.json").is_file() and (repo / "names" / "alice.id.json").is_file()+ assert "from the GitHub account" not in out # a network's repository asks no GitHub id+ shutil.copytree(repo / "names", tmp_path / "base")+ # Moving the name to a new key: the old key signs the handover.+ run(capsys, "id", "new", "--dir", new)+ (repo / "names" / "alice.id.json").unlink()+ code, _, err = run(capsys, "id", "name", f"alice@{DOMAIN}", "--into", str(repo), "--handover-from", str(tmp_path / "me" / "root.key"), "--dir", new)+ assert code == 0, err+ code, out, _ = run(capsys, "names", "gate", str(repo / "names"), "--base", str(tmp_path / "base"), "--domain", DOMAIN)+ assert code == 0 and "open" in out+ # Giving it up: a release in its place.+ code, _, _ = run(capsys, "id", "release", f"alice@{DOMAIN}", "--out", str(repo / "names" / "alice.json"), "--dir", new)+ (repo / "names" / "alice.id.json").unlink()+ code, out, _ = run(capsys, "id", "check", str(repo / "names" / "alice.json"))+ assert code == 0 and "gives up" in out+ code, out, _ = run(capsys, "names", "gate", str(repo / "names"), "--base", str(tmp_path / "base"), "--domain", DOMAIN)+ assert code == 1 and "is held by" in out # the base still holds it under the first key