/
/
/
1"""Server identity persistence for the Sendspin provider."""
2
3from __future__ import annotations
4
5import os
6from pathlib import Path
7
8from aiosendspin.noise.keys import Identity, b64url_decode
9
10IDENTITY_FILENAME = "identity.key"
11
12
13def get_or_create_server_identity(storage_dir: Path) -> Identity:
14 """
15 Load the persistent Sendspin server identity, generating one only if absent.
16
17 Raises on a corrupt or unreadable key file rather than minting a new identity
18 over trusted state. Blocking (file I/O) - call via asyncio.to_thread.
19
20 :param storage_dir: Directory holding the identity key file (created if needed).
21 """
22 storage_dir.mkdir(parents=True, exist_ok=True)
23 key_path = storage_dir / IDENTITY_FILENAME
24 try:
25 return Identity.from_private_bytes(b64url_decode(key_path.read_text().strip()))
26 except FileNotFoundError:
27 pass
28 identity = Identity.generate()
29 try:
30 fd = os.open(key_path, os.O_CREAT | os.O_WRONLY | os.O_EXCL, 0o600)
31 except FileExistsError:
32 return Identity.from_private_bytes(b64url_decode(key_path.read_text().strip()))
33 with os.fdopen(fd, "w") as f:
34 f.write(identity.private_b64u)
35 return identity
36