pyrtan / sourcev0.19.0 · main

pyrtan / src / pyrtan / identity.py

32,659 bytes · last changed 2026-10-28 in identity.js: the documents a root signs, and encrypted backups, agreeing with Python

1"""Identity: a person, a hub or a device as an Ed25519 key.
2
3An identity is its public key, written ``pyr:`` and 56 characters: the
4lowercase base32 of the 32-byte key, a 2-byte SHA3-256 checksum and a
5version byte, the layout of a Tor v3 onion address with Pyrtan's own
6checksum string. ``pyr:<key>@host`` says where an identity can be
7reached; ``alice@example.org`` is a name a domain vouches for. Encoding,
8decoding and parsing those need only the standard library.
9
10Three kinds of signed document stand on their own, so that anyone can
11hold a copy and nobody has to trust the holder: the *identity document*
12(routes, devices, revocations), the *device note* (a root delegating to
13a device key, with what it may do and until when), and the *name claim*
14(an identity saying it is ``name@domain``). Each is signed over
15``pyrtan-doc-v1\\n`` and the document's canonical JSON without ``sig``.
16
17Signing and verifying need ``cryptography`` (``pip install
18'pyrtan[identity]'``), imported only when a signature is made or
19checked. This module refuses to do Ed25519 itself: a private key is
20handled only by a maintained library. It also refuses floats in a
21document, so that every runtime produces the same bytes. The design is
22``_meta/identity.md``.
23"""
24
25from __future__ import annotations
26
27import base64
28import hashlib
29import ipaddress
30import json
31import os
32import re
33import time
34from dataclasses import dataclass
35from pathlib import Path
36from typing import Any
37
38from .store import canonical
39
40PREFIX = "pyr:"
41VERSION_ED25519 = 1
42VERSIONS = {VERSION_ED25519: 32}
43CHECKSUM_TAG = b".pyrtan id"
44DOC_TAG = b"pyrtan-doc-v1\n"
45REQUEST_TAG = "pyrtan-req-v1"
46BACKUP_TAG = b"pyrtan-key-backup-v1\n"
47HANDOVER_TAG = b"pyrtan-handover-v1\n"
48BACKUP_ITERATIONS = 600_000 # PBKDF2-SHA256, as OWASP advises for it in 2023 and after
49MIN_ITERATIONS = 100_000 # what a backup may claim, at the least, and still be read
50ID_LENGTH = 56
51WINDOW = 300 # seconds a document's time may run ahead of ours
52YEAR = 365 * 24 * 3600 # the longest a device note may last
53ABILITIES = ("ask", "chat", "run", "operate")
54KINDS = ("person", "hub")
55MAX_ROUTES = 16
56MAX_DEVICES = 64
57
58LOCAL = re.compile(r"^[a-z0-9][a-z0-9._-]{0,62}$")
59LABEL = re.compile(r"^[a-z0-9]([a-z0-9-]{0,61}[a-z0-9])?$")
60INSTALL = "signatures need the identity extra: pip install 'pyrtan[identity]'"
61
62
63class IdentityError(ValueError):
64 pass
65
66
67# ---------------------------------------------------------------- identifiers
68
69
70def _checksum(key: bytes, version: int) -> bytes:
71 return hashlib.sha3_256(CHECKSUM_TAG + key + bytes([version])).digest()[:2]
72
73
74def encode(key: bytes, version: int = VERSION_ED25519) -> str:
75 """The identifier of a public key."""
76 if VERSIONS.get(version) != len(key):
77 raise IdentityError(f"a version {version} key is {VERSIONS.get(version, '?')} bytes, not {len(key)}")
78 raw = key + _checksum(key, version) + bytes([version])
79 return PREFIX + base64.b32encode(raw).decode("ascii").lower().rstrip("=")
80
81
82def decode(text: Any) -> bytes:
83 """The public key an identifier names; IdentityError says what is wrong."""
84 if not isinstance(text, str) or not text.lower().startswith(PREFIX):
85 raise IdentityError(f"{text!r} is not an identity: it starts {PREFIX}")
86 body = text[len(PREFIX):].lower()
87 if len(body) != ID_LENGTH:
88 raise IdentityError(f"{text!r} is not an identity: {len(body)} characters after {PREFIX}, not {ID_LENGTH}")
89 try:
90 raw = base64.b32decode(body.upper() + "=" * (-len(body) % 8))
91 except ValueError:
92 raise IdentityError(f"{text!r} is not an identity: not base32") from None
93 key, checksum, version = raw[:-3], raw[-3:-1], raw[-1]
94 if version not in VERSIONS:
95 raise IdentityError(f"{text!r} is version {version}, which this pyrtan does not know")
96 if _checksum(key, version) != checksum:
97 raise IdentityError(f"{text!r} is not an identity: the checksum does not match (a typo?)")
98 return key
99
100
101def normal(text: Any) -> str:
102 """The identifier as it is compared: checked, lowercase."""
103 decode(text)
104 return text.lower()
105
106
107def short(text: str) -> str:
108 """For display only; never compared."""
109 return f"{PREFIX}{text[len(PREFIX):len(PREFIX) + 4]}…{text[-4:]}"
110
111
112def check_locator(text: Any) -> str:
113 """``host``, ``host:port``, ``<56>.onion`` or ``[ip]:port``, lowercased."""
114 if not isinstance(text, str) or not text:
115 raise IdentityError(f"{text!r} is not a locator")
116 loc = text.lower()
117 host, port = loc, None
118 if loc.startswith("["):
119 end = loc.find("]")
120 if end < 0:
121 raise IdentityError(f"{text!r} is not a locator: an unclosed [")
122 host, rest = loc[1:end], loc[end + 1:]
123 try:
124 ipaddress.IPv6Address(host)
125 except ValueError:
126 raise IdentityError(f"{text!r} is not a locator: {host!r} is not an IPv6 address") from None
127 if rest:
128 if not rest.startswith(":"):
129 raise IdentityError(f"{text!r} is not a locator")
130 port = rest[1:]
131 else:
132 if ":" in loc:
133 host, port = loc.rsplit(":", 1)
134 check_host(host, text)
135 if port is not None and (not port.isdigit() or not 0 < int(port) < 65536):
136 raise IdentityError(f"{text!r} is not a locator: {port!r} is not a port")
137 return loc
138
139
140def check_host(host: str, given: Any = None) -> str:
141 given = host if given is None else given
142 labels = host.split(".")
143 if len(host) > 253 or not all(LABEL.match(label) for label in labels):
144 raise IdentityError(f"{given!r} is not a locator: {host!r} is not a host name")
145 if labels[-1] == "onion" and (len(labels) < 2 or len(labels[-2]) != 56):
146 raise IdentityError(f"{given!r} is not a locator: an onion address is 56 characters before .onion")
147 return host
148
149
150@dataclass(frozen=True)
151class Parsed:
152 form: str # "identity", "address" or "name"
153 id: str | None = None
154 locator: str | None = None
155 name: str | None = None
156 domain: str | None = None
157
158 def __str__(self) -> str:
159 if self.form == "name":
160 return f"{self.name}@{self.domain}"
161 return self.id + (f"@{self.locator}" if self.locator else "")
162
163
164def parse(text: Any) -> Parsed:
165 """Any of the three forms: ``pyr:<key>``, ``pyr:<key>@locator``, ``name@domain``."""
166 if not isinstance(text, str) or not text.strip():
167 raise IdentityError(f"{text!r} is not an identity, an address or a name")
168 text = text.strip()
169 if text.lower().startswith(PREFIX):
170 ident, _, loc = text.partition("@")
171 ident = normal(ident)
172 if not _:
173 return Parsed("identity", id=ident)
174 return Parsed("address", id=ident, locator=check_locator(loc))
175 local, at, domain = text.lower().rpartition("@")
176 if not at:
177 raise IdentityError(f"{text!r} is not an identity ({PREFIX}…) or a name (name@domain)")
178 if not LOCAL.match(local):
179 raise IdentityError(f"{text!r} is not a name: {local!r} is not lowercase letters, digits, dot, dash, underscore, at most 63")
180 return Parsed("name", name=local, domain=check_host(domain, text))
181
182
183# ---------------------------------------------------------------- signatures
184
185
186def _ed25519():
187 try:
188 from cryptography.hazmat.primitives.asymmetric import ed25519
189 except ImportError:
190 raise IdentityError(INSTALL) from None
191 return ed25519
192
193
194def require_signatures() -> None:
195 """IdentityError, saying how to install the extra, if this pyrtan cannot sign or verify."""
196 _ed25519()
197
198
199def _b64(raw: bytes) -> str:
200 return base64.urlsafe_b64encode(raw).decode("ascii").rstrip("=")
201
202
203def _unb64(text: Any) -> bytes:
204 if not isinstance(text, str) or not re.fullmatch(r"[A-Za-z0-9_-]*", text):
205 raise IdentityError("not base64url")
206 return base64.urlsafe_b64decode(text + "=" * (-len(text) % 4))
207
208
209class Key:
210 """A private key, held in memory; ``id`` is its identity."""
211
212 def __init__(self, seed: bytes):
213 if len(seed) != 32:
214 raise IdentityError("an Ed25519 private key is 32 bytes")
215 self._private = _ed25519().Ed25519PrivateKey.from_private_bytes(seed)
216 self._seed = seed
217 from cryptography.hazmat.primitives import serialization as s
218 self.id = encode(self._private.public_key().public_bytes(s.Encoding.Raw, s.PublicFormat.Raw))
219
220 @classmethod
221 def generate(cls) -> "Key":
222 return cls(os.urandom(32))
223
224 def sign(self, message: bytes) -> bytes:
225 return self._private.sign(message)
226
227 def __repr__(self) -> str:
228 return f"Key({short(self.id)})"
229
230 # A key file is JSON, written once, readable only by its owner.
231
232 def save(self, path: Path, *, kind: str = "root") -> Path:
233 path.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
234 data = json.dumps({"pyrtan-key": 1, "kind": kind, "id": self.id, "seed": _b64(self._seed)}, indent=2) + "\n"
235 fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
236 with os.fdopen(fd, "w", encoding="utf-8") as f:
237 f.write(data)
238 return path
239
240 @classmethod
241 def load(cls, path: Path) -> "Key":
242 try:
243 data = json.loads(path.read_text(encoding="utf-8"))
244 except (OSError, json.JSONDecodeError) as e:
245 raise IdentityError(f"{path}: {e}") from None
246 if not isinstance(data, dict) or data.get("pyrtan-key") != 1:
247 raise IdentityError(f"{path} is not a pyrtan key file")
248 key = cls(_unb64(data.get("seed")))
249 if data.get("id") != key.id:
250 raise IdentityError(f"{path}: the id in the file is not the key's")
251 return key
252
253
254def verify(ident: str, message: bytes, signature: bytes) -> bool:
255 """Whether ``signature`` is ``ident``'s over ``message``."""
256 public = _ed25519().Ed25519PublicKey.from_public_bytes(decode(ident))
257 try:
258 public.verify(signature, message)
259 except Exception:
260 return False
261 return True
262
263
264# ---------------------------------------------------------------- documents
265
266
267def _plain(value: Any, where: str = "document") -> None:
268 if isinstance(value, float):
269 raise IdentityError(f"{where} holds a number with a fraction; documents hold strings, integers, lists and objects")
270 if isinstance(value, dict):
271 for k, v in value.items():
272 if not isinstance(k, str):
273 raise IdentityError(f"{where} has a key that is not a string")
274 _plain(v, f"{where}.{k}")
275 elif isinstance(value, list):
276 for i, v in enumerate(value):
277 _plain(v, f"{where}[{i}]")
278 elif not (value is None or isinstance(value, (str, int, bool))):
279 raise IdentityError(f"{where} holds a {type(value).__name__}")
280
281
282def signed_bytes(doc: dict[str, Any]) -> bytes:
283 """What a document's ``sig`` is over."""
284 body = {k: v for k, v in doc.items() if k != "sig"}
285 _plain(body)
286 return DOC_TAG + canonical(body)
287
288
289def sign_document(doc: dict[str, Any], key: Key) -> dict[str, Any]:
290 body = {k: v for k, v in doc.items() if k != "sig"}
291 return {**body, "sig": _b64(key.sign(signed_bytes(body)))}
292
293
294def check_signature(doc: Any, by: str) -> str | None:
295 """None if ``doc`` is signed by ``by``; otherwise what is wrong."""
296 if not isinstance(doc, dict) or "sig" not in doc:
297 return "not signed"
298 try:
299 signature = _unb64(doc["sig"])
300 message = signed_bytes(doc)
301 except IdentityError as e:
302 return str(e)
303 except ValueError:
304 return "the signature is not base64url"
305 return None if verify(by, message, signature) else f"the signature is not {short(by)}'s"
306
307
308def _int(doc: dict[str, Any], field: str) -> int:
309 value = doc.get(field)
310 if not isinstance(value, int) or isinstance(value, bool) or value < 0:
311 raise IdentityError(f"{field} is not a time (whole seconds since 1970)")
312 return value
313
314
315def _abilities(value: Any) -> list[str]:
316 if not isinstance(value, list) or not all(a in ABILITIES for a in value) or len(set(value)) != len(value):
317 raise IdentityError(f"may is a list of {', '.join(ABILITIES)}, each once")
318 return value
319
320
321# Device notes.
322
323
324def make_device(root: Key, device: str, may: list[str], *, label: str = "", issued: int | None = None, expires: int | None = None) -> dict[str, Any]:
325 issued = int(time.time()) if issued is None else issued
326 note = {
327 "pyrtan-device": 1, "root": root.id, "device": normal(device), "label": label,
328 "may": [x for x in ABILITIES if x in _abilities(may)], "issued": issued, "expires": issued + YEAR if expires is None else expires,
329 }
330 if note["device"] == root.id:
331 raise IdentityError("a root does not delegate to itself; it signs for itself")
332 _check_device_fields(note)
333 return sign_document(note, root)
334
335
336def _check_device_fields(note: dict[str, Any]) -> None:
337 if note.get("pyrtan-device") != 1:
338 raise IdentityError("not a device note (pyrtan-device: 1)")
339 normal(note.get("root"))
340 normal(note.get("device"))
341 if not isinstance(note.get("label", ""), str) or len(note.get("label", "")) > 64:
342 raise IdentityError("label is a string of at most 64 characters")
343 _abilities(note.get("may"))
344 issued, expires = _int(note, "issued"), _int(note, "expires")
345 if not issued < expires <= issued + YEAR:
346 raise IdentityError("a device note expires after it is issued, and within a year")
347
348
349def check_device(note: Any, *, root: str | None = None, now: float | None = None) -> str | None:
350 """None if the note holds (signed by its root, and by ``root`` if given, and current); otherwise why not."""
351 if not isinstance(note, dict):
352 return "not a device note"
353 try:
354 _check_device_fields(note)
355 except IdentityError as e:
356 return str(e)
357 if root is not None and note["root"].lower() != root.lower():
358 return f"the note is {short(note['root'])}'s, not {short(root)}'s"
359 problem = check_signature(note, note["root"])
360 if problem:
361 return problem
362 now = time.time() if now is None else now
363 if note["issued"] > now + WINDOW:
364 return "the note is issued in the future"
365 if note["expires"] <= now:
366 return "the note has expired"
367 return None
368
369
370# Identity documents.
371
372
373def make_identity(root: Key, *, kind: str = "person", routes: list[str] | None = None, devices: list[dict[str, Any]] | None = None,
374 revoked: list[str] | None = None, updated: int | None = None) -> dict[str, Any]:
375 doc = {
376 "pyrtan-identity": 1, "id": root.id, "kind": kind,
377 "routes": [check_locator(r) for r in routes or []],
378 "devices": list(devices or []), "revoked": [normal(r) for r in revoked or []],
379 "updated": int(time.time()) if updated is None else updated,
380 }
381 _check_identity_fields(doc)
382 return sign_document(doc, root)
383
384
385def _check_identity_fields(doc: dict[str, Any]) -> None:
386 if doc.get("pyrtan-identity") != 1:
387 raise IdentityError("not an identity document (pyrtan-identity: 1)")
388 ident = normal(doc.get("id"))
389 if doc.get("kind") not in KINDS:
390 raise IdentityError(f"kind is one of {', '.join(KINDS)}")
391 routes = doc.get("routes")
392 if not isinstance(routes, list) or len(routes) > MAX_ROUTES:
393 raise IdentityError(f"routes is a list of at most {MAX_ROUTES} locators")
394 for r in routes:
395 if check_locator(r) != r:
396 raise IdentityError(f"route {r!r} is not lowercase")
397 revoked = doc.get("revoked")
398 if not isinstance(revoked, list) or any(normal(r) != r for r in revoked):
399 raise IdentityError("revoked is a list of lowercase identities")
400 devices = doc.get("devices")
401 if not isinstance(devices, list) or len(devices) > MAX_DEVICES:
402 raise IdentityError(f"devices is a list of at most {MAX_DEVICES} device notes")
403 seen = set()
404 for note in devices:
405 if not isinstance(note, dict):
406 raise IdentityError("devices holds something that is not a device note")
407 _check_device_fields(note)
408 if note["root"] != ident:
409 raise IdentityError(f"device {short(note['device'])} is delegated by {short(note['root'])}, not this identity")
410 if note["device"] in seen:
411 raise IdentityError(f"device {short(note['device'])} is listed twice")
412 seen.add(note["device"])
413 _int(doc, "updated")
414
415
416def check_identity(doc: Any, *, ident: str | None = None, now: float | None = None) -> str | None:
417 """None if the document holds (well formed, signed by its own id, every note signed by it, not from the future)."""
418 if not isinstance(doc, dict):
419 return "not an identity document"
420 try:
421 _check_identity_fields(doc)
422 except IdentityError as e:
423 return str(e)
424 if ident is not None and doc["id"] != ident.lower():
425 return f"the document is {short(doc['id'])}'s, not {short(ident)}'s"
426 problem = check_signature(doc, doc["id"])
427 if problem:
428 return problem
429 for note in doc["devices"]:
430 problem = check_signature(note, doc["id"])
431 if problem:
432 return f"device {short(note['device'])}: {problem}"
433 if doc["updated"] > (time.time() if now is None else now) + WINDOW:
434 return "the document is from the future"
435 return None
436
437
438def newest(docs: list[Any], *, ident: str | None = None, now: float | None = None) -> dict[str, Any] | None:
439 """Of several copies, the one with the highest ``updated`` whose checks hold."""
440 good = [d for d in docs if check_identity(d, ident=ident, now=now) is None]
441 return max(good, key=lambda d: d["updated"], default=None)
442
443
444def device_status(doc: dict[str, Any], device: str, *, now: float | None = None) -> str:
445 """``ok``, ``expired``, ``revoked`` or ``unknown``, for a device under a checked identity document."""
446 device = device.lower()
447 if device in doc["revoked"]:
448 return "revoked"
449 for note in doc["devices"]:
450 if note["device"] == device:
451 return "ok" if note["expires"] > (time.time() if now is None else now) else "expired"
452 return "unknown"
453
454
455def with_device(doc: dict[str, Any], root: Key, note: dict[str, Any]) -> dict[str, Any]:
456 """A new document, signed, with ``note`` in place of any earlier note for the same device."""
457 devices = [n for n in doc["devices"] if n["device"] != note["device"]] + [note]
458 revoked = [r for r in doc["revoked"] if r != note["device"]]
459 return _reissue(doc, root, devices=devices, revoked=revoked)
460
461
462def with_revoked(doc: dict[str, Any], root: Key, device: str) -> dict[str, Any]:
463 device = normal(device)
464 devices = [n for n in doc["devices"] if n["device"] != device]
465 revoked = doc["revoked"] + ([device] if device not in doc["revoked"] else [])
466 return _reissue(doc, root, devices=devices, revoked=revoked)
467
468
469def with_routes(doc: dict[str, Any], root: Key, routes: list[str]) -> dict[str, Any]:
470 return _reissue(doc, root, routes=routes)
471
472
473def _reissue(doc: dict[str, Any], root: Key, **changes: Any) -> dict[str, Any]:
474 if doc["id"] != root.id:
475 raise IdentityError(f"the document is {short(doc['id'])}'s; this key is {short(root.id)}")
476 fields = {k: doc[k] for k in ("kind", "routes", "devices", "revoked")} | changes
477 # Strictly newer than the copy it replaces, or a holder would keep the old one.
478 return make_identity(root, updated=max(int(time.time()), doc["updated"] + 1), **fields)
479
480
481# Name claims.
482
483
484def make_name(key: Key, name: str, *, issued: int | None = None, handover_from: Key | None = None, **extra: Any) -> dict[str, Any]:
485 """``key`` claiming ``name`` (``local@domain``). ``extra`` is a domain's own fields (pyrtan.com's ``github``).
486 ``handover_from`` is the key that holds the name now: it signs the claim's body first, as
487 ``handover``, which is how a name moves to a new key where names are owned by their key."""
488 parsed = parse(name)
489 if parsed.form != "name":
490 raise IdentityError(f"{name!r} is not a name (name@domain)")
491 claim = {"pyrtan-name": 1, "name": parsed.name, "domain": parsed.domain, "id": key.id,
492 "issued": int(time.time()) if issued is None else issued, **extra}
493 _plain(claim)
494 if handover_from is not None:
495 claim["handover"] = {"id": handover_from.id, "sig": _b64(handover_from.sign(HANDOVER_TAG + canonical(claim)))}
496 return sign_document(claim, key)
497
498
499def check_name(claim: Any, *, name: str | None = None) -> str | None:
500 """None if the claim holds, and is for ``name`` if given; otherwise why not."""
501 if not isinstance(claim, dict) or claim.get("pyrtan-name") != 1:
502 return "not a name claim (pyrtan-name: 1)"
503 try:
504 normal(claim.get("id"))
505 _int(claim, "issued")
506 stated = parse(f"{claim.get('name')}@{claim.get('domain')}")
507 except IdentityError as e:
508 return str(e)
509 if stated.form != "name" or (stated.name, stated.domain) != (claim["name"], claim["domain"]):
510 return "name and domain are not a lowercase name@domain"
511 if name is not None and str(parse(name)) != str(stated):
512 return f"the claim is for {stated}, not {name}"
513 return check_signature(claim, claim["id"])
514
515
516def check_handover(claim: dict[str, Any], old: str) -> str | None:
517 """None if ``claim`` carries ``old``'s handover; otherwise why not."""
518 h = claim.get("handover")
519 if not isinstance(h, dict) or not isinstance(h.get("sig"), str):
520 return f"the name moves from {short(old)} to {short(claim.get('id', '?'))} without a handover signed by {short(old)}"
521 if h.get("id") != old:
522 return f"the handover is {short(str(h.get('id')))}'s, not {short(old)}'s"
523 body = {k: v for k, v in claim.items() if k not in ("sig", "handover")}
524 try:
525 ok = verify(old, HANDOVER_TAG + canonical(body), _unb64(h["sig"]))
526 except (IdentityError, ValueError):
527 ok = False
528 return None if ok else f"the handover is not signed by {short(old)}"
529
530
531def make_release(key: Key, name: str, *, issued: int | None = None) -> dict[str, Any]:
532 """``key`` giving up ``name``: what replaces its claim, so the name is free again."""
533 parsed = parse(name)
534 if parsed.form != "name":
535 raise IdentityError(f"{name!r} is not a name (name@domain)")
536 release = {"pyrtan-name-release": 1, "name": parsed.name, "domain": parsed.domain, "id": key.id,
537 "issued": int(time.time()) if issued is None else issued}
538 return sign_document(release, key)
539
540
541def check_release(release: Any, *, name: str | None = None) -> str | None:
542 if not isinstance(release, dict) or release.get("pyrtan-name-release") != 1:
543 return "not a name release (pyrtan-name-release: 1)"
544 try:
545 normal(release.get("id"))
546 _int(release, "issued")
547 stated = parse(f"{release.get('name')}@{release.get('domain')}")
548 except IdentityError as e:
549 return str(e)
550 if stated.form != "name" or (stated.name, stated.domain) != (release["name"], release["domain"]):
551 return "name and domain are not a lowercase name@domain"
552 if name is not None and str(parse(name)) != str(stated):
553 return f"the release is for {stated}, not {name}"
554 return check_signature(release, release["id"])
555
556
557def check_any(doc: Any, *, now: float | None = None) -> tuple[str, str | None]:
558 """Which kind of document this is, and None or what is wrong with it."""
559 if isinstance(doc, dict):
560 if "pyrtan-identity" in doc:
561 return "identity", check_identity(doc, now=now)
562 if "pyrtan-device" in doc:
563 return "device", check_device(doc, now=now)
564 if "pyrtan-name" in doc:
565 return "name", check_name(doc)
566 if "pyrtan-name-release" in doc:
567 return "release", check_release(doc)
568 if "pyrtan-key-backup" in doc:
569 return "backup", backup_problem(doc)
570 return "unknown", "not a pyrtan identity document, device note, name claim, release or key backup"
571
572
573# ---------------------------------------------------------------- signed requests
574#
575# A request made as an identity carries X-Pyrtan-Key (the signing key, a
576# root or a device), X-Pyrtan-Date, X-Pyrtan-Key-Signature (Ed25519 over the
577# message below) and, for a device, X-Pyrtan-Device (its note, as base64url
578# canonical JSON). The message is the HMAC one with a domain string and the
579# *audience*, the receiving hub's own id, in front: a request signed for one
580# hub cannot be replayed at another.
581
582
583def request_message(audience: str, method: str, path: str, date: str, body: Any) -> bytes:
584 digest = hashlib.sha256(canonical(body) if body is not None else b"").hexdigest()
585 return "\n".join([REQUEST_TAG, audience.rstrip("/"), method.upper(), path, date, digest]).encode("utf-8")
586
587
588def sign_request(key: Key, audience: str, method: str, path: str, body: Any, *, note: dict[str, Any] | None = None,
589 date: float | None = None) -> dict[str, str]:
590 """The headers that make a request ``key``'s; ``note`` when ``key`` is a device."""
591 date_text = str(int(time.time() if date is None else date))
592 headers = {
593 "X-Pyrtan-Key": key.id, "X-Pyrtan-Date": date_text,
594 "X-Pyrtan-Key-Signature": _b64(key.sign(request_message(audience, method, path, date_text, body))),
595 }
596 if note is not None:
597 if note.get("device") != key.id:
598 raise IdentityError(f"the note is for {short(note.get('device', '?'))}, not this key")
599 headers["X-Pyrtan-Device"] = _b64(canonical(note))
600 return headers
601
602
603@dataclass(frozen=True)
604class Signer:
605 """Who signed a request: the key, the root it speaks for, and the device note if it is a device."""
606 key: str
607 root: str
608 note: dict[str, Any] | None = None
609
610
611def has_key_signature(headers: dict[str, str] | None) -> bool:
612 return any(k.lower() == "x-pyrtan-key-signature" for k in (headers or {}))
613
614
615def verify_request(audience: str, method: str, path: str, body: Any, headers: dict[str, str] | None, *,
616 now: float | None = None) -> Signer:
617 """The signer of a request, or IdentityError saying why there is none. Checks the
618 signature, the time, and the device note (signed by its root, current, for this key);
619 whether the root may do anything is the caller's question."""
620 h = {k.lower(): v for k, v in (headers or {}).items()}
621 key, date, signature = h.get("x-pyrtan-key", ""), h.get("x-pyrtan-date", ""), h.get("x-pyrtan-key-signature", "")
622 if not key or not signature:
623 raise IdentityError("no key signature")
624 key = normal(key)
625 if not date.isdigit():
626 raise IdentityError("no date")
627 now = time.time() if now is None else now
628 if abs(now - int(date)) > WINDOW:
629 raise IdentityError("signature too old")
630 try:
631 raw = _unb64(signature)
632 except (IdentityError, ValueError):
633 raise IdentityError("the signature is not base64url") from None
634 if not verify(key, request_message(audience, method, path, date, body), raw):
635 raise IdentityError(f"the signature is not {short(key)}'s, for {audience}")
636 if "x-pyrtan-device" not in h:
637 return Signer(key, key)
638 try:
639 note = json.loads(_unb64(h["x-pyrtan-device"]).decode("utf-8"))
640 except (IdentityError, ValueError):
641 raise IdentityError("X-Pyrtan-Device is not a base64url note") from None
642 problem = check_device(note, now=now)
643 if problem:
644 raise IdentityError(f"device note: {problem}")
645 if note["device"] != key:
646 raise IdentityError(f"the device note is for {short(note['device'])}, not the signing key")
647 return Signer(key, note["root"], note)
648
649
650# ---------------------------------------------------------------- backups
651#
652# A root key, encrypted with a passphrase, beside the identity document it
653# signs, so that a person can keep it somewhere durable (a file, their own
654# private repository) and restore it on another machine or browser. The
655# console writes the same format with WebCrypto (identity.js). It is for
656# places only its owner can read: an encrypted key where anyone can copy
657# it can be guessed at offline, forever.
658
659
660def _aesgcm_kdf():
661 try:
662 from cryptography.hazmat.primitives import hashes
663 from cryptography.hazmat.primitives.ciphers.aead import AESGCM
664 from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2HMAC
665 except ImportError:
666 raise IdentityError(INSTALL) from None
667 return hashes, AESGCM, PBKDF2HMAC
668
669
670def _backup_key(passphrase: str, salt: bytes, iterations: int) -> bytes:
671 hashes, _, PBKDF2HMAC = _aesgcm_kdf()
672 return PBKDF2HMAC(algorithm=hashes.SHA256(), length=32, salt=salt, iterations=iterations).derive(passphrase.encode("utf-8"))
673
674
675def make_backup(key: Key, passphrase: str, *, document: dict[str, Any] | None = None, salt: bytes | None = None,
676 iv: bytes | None = None, iterations: int = BACKUP_ITERATIONS, made: int | None = None) -> dict[str, Any]:
677 """``key`` encrypted with ``passphrase`` (PBKDF2-SHA256, then AES-256-GCM over the 32-byte
678 seed, the identifier as associated data), with its identity document beside it."""
679 if len(passphrase) < 8:
680 raise IdentityError("a passphrase is at least 8 characters; a long sentence is better")
681 _, AESGCM, _ = _aesgcm_kdf()
682 salt = os.urandom(16) if salt is None else salt
683 iv = os.urandom(12) if iv is None else iv
684 data = AESGCM(_backup_key(passphrase, salt, iterations)).encrypt(iv, key._seed, BACKUP_TAG + key.id.encode("ascii"))
685 backup = {
686 "pyrtan-key-backup": 1, "id": key.id, "made": int(time.time()) if made is None else made,
687 "kdf": {"name": "PBKDF2-SHA256", "iterations": iterations, "salt": _b64(salt)},
688 "cipher": {"name": "AES-256-GCM", "iv": _b64(iv)}, "data": _b64(data),
689 }
690 if document is not None:
691 if document.get("id") != key.id:
692 raise IdentityError("the document is another identity's")
693 backup["document"] = document
694 return backup
695
696
697def backup_problem(backup: Any) -> str | None:
698 """What is wrong with a backup's shape, before any passphrase is tried; None if nothing."""
699 if not isinstance(backup, dict) or backup.get("pyrtan-key-backup") != 1:
700 return "not a key backup (pyrtan-key-backup: 1)"
701 kdf, cipher = backup.get("kdf"), backup.get("cipher")
702 try:
703 normal(backup.get("id"))
704 except IdentityError as e:
705 return str(e)
706 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):
707 return "the backup's kdf is not PBKDF2-SHA256 with iterations and a salt"
708 if not MIN_ITERATIONS <= kdf["iterations"] <= 10_000_000:
709 return f"the backup's kdf claims {kdf['iterations']} iterations; at least {MIN_ITERATIONS} are read"
710 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):
711 return "the backup's cipher is not AES-256-GCM with an iv and data"
712 doc = backup.get("document")
713 if doc is not None:
714 problem = check_identity(doc, ident=backup["id"])
715 if problem:
716 return f"the document beside the key: {problem}"
717 return None
718
719
720def open_backup(backup: Any, passphrase: str) -> Key:
721 """The key in ``backup``; IdentityError for a wrong passphrase or a damaged file."""
722 problem = backup_problem(backup)
723 if problem:
724 raise IdentityError(problem)
725 _, AESGCM, _ = _aesgcm_kdf()
726 try:
727 salt, iv, data = _unb64(backup["kdf"]["salt"]), _unb64(backup["cipher"]["iv"]), _unb64(backup["data"])
728 seed = AESGCM(_backup_key(passphrase, salt, backup["kdf"]["iterations"])).decrypt(iv, data, BACKUP_TAG + backup["id"].lower().encode("ascii"))
729 except Exception:
730 raise IdentityError("the passphrase does not open this backup (or the file is damaged)") from None
731 key = Key(seed)
732 if key.id != backup["id"].lower():
733 raise IdentityError("the backup's key is not the identity it names")
734 return key
735
736
737# ---------------------------------------------------------------- where the CLI keeps them
738
739
740def home() -> Path:
741 """``$PYRTAN_IDENTITY``, else ``$XDG_CONFIG_HOME/pyrtan/identity``, else ``~/.config/pyrtan/identity``."""
742 if os.environ.get("PYRTAN_IDENTITY"):
743 return Path(os.environ["PYRTAN_IDENTITY"])
744 base = Path(os.environ.get("XDG_CONFIG_HOME") or Path.home() / ".config")
745 return base / "pyrtan" / "identity"