1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
|
"""
MeshBay Hub — authentication helpers.
- Password hashing/verification: Argon2id
- JWT issuance/verification: Ed25519 (EdDSA), includes jti
- Refresh token: random 32-byte, stored as blake3 hex hash
- Hub keypair: loaded from PEM file on startup
"""
import base64
import hashlib
import os
import time
import uuid
from pathlib import Path
import blake3
import jwt
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.primitives.kdf.argon2 import Argon2id
# Argon2id parameters — see CLAUDE.md for production calibration guidance
_ARGON2_ITERATIONS = 3
_ARGON2_MEMORY_COST = 65536 # 64 MB — increase to 262144 in production
_ARGON2_LANES = 4
_ARGON2_KEY_LEN = 32
# Module-level hub keypair (loaded once at startup)
_hub_sk_pem: bytes | None = None
_hub_pk_pem: bytes | None = None
_hub_id: str = "meshbay.org"
# ── Hub keypair ───────────────────────────────────────────────────────────────
def load_hub_keypair(private_key_path: Path, hub_id: str) -> None:
"""Load hub Ed25519 keypair from PEM file. Call once at startup."""
global _hub_sk_pem, _hub_pk_pem, _hub_id
_hub_sk_pem = private_key_path.read_bytes()
sk = serialization.load_pem_private_key(_hub_sk_pem, password=None)
_hub_pk_pem = sk.public_key().public_bytes(
serialization.Encoding.PEM,
serialization.PublicFormat.SubjectPublicKeyInfo,
)
_hub_id = hub_id
def generate_hub_keypair(private_key_path: Path) -> None:
"""Generate a new hub Ed25519 keypair and save PEM files. Run once."""
private_key_path.parent.mkdir(parents=True, exist_ok=True)
sk = Ed25519PrivateKey.generate()
private_key_path.write_bytes(sk.private_bytes(
serialization.Encoding.PEM,
serialization.PrivateFormat.PKCS8,
serialization.NoEncryption(),
))
private_key_path.chmod(0o600)
pk_path = private_key_path.with_suffix(".pub.pem")
pk_path.write_bytes(sk.public_key().public_bytes(
serialization.Encoding.PEM,
serialization.PublicFormat.SubjectPublicKeyInfo,
))
def hub_public_key_pem() -> bytes:
if _hub_pk_pem is None:
raise RuntimeError("Hub keypair not loaded — call load_hub_keypair() first")
return _hub_pk_pem
# ── Password ──────────────────────────────────────────────────────────────────
def hash_password(password: str) -> tuple[bytes, bytes]:
"""Hash a password with Argon2id. Returns (hash, salt)."""
salt = os.urandom(16)
pw_hash = Argon2id(
salt=salt,
length=_ARGON2_KEY_LEN,
iterations=_ARGON2_ITERATIONS,
lanes=_ARGON2_LANES,
memory_cost=_ARGON2_MEMORY_COST,
).derive(password.encode())
return pw_hash, salt
def verify_password(password: str, pw_hash: bytes, salt: bytes) -> bool:
try:
Argon2id(
salt=salt,
length=_ARGON2_KEY_LEN,
iterations=_ARGON2_ITERATIONS,
lanes=_ARGON2_LANES,
memory_cost=_ARGON2_MEMORY_COST,
).verify(password.encode(), pw_hash)
return True
except Exception:
return False
# ── JWT ───────────────────────────────────────────────────────────────────────
def issue_access_token(
user_id: str,
pk_user: str,
ttl: int = 3600,
groups: list[str] | None = None,
) -> str:
"""
Issue a signed JWT access token.
Includes jti (UUID4) — required to prevent replay and enable revocation.
Includes groups — list of group_ids the user is a member of (node-side authz).
"""
if _hub_sk_pem is None:
raise RuntimeError("Hub keypair not loaded")
now = int(time.time())
payload = {
"iss": _hub_id,
"sub": user_id,
"pk_user": pk_user,
"hub_id": _hub_id,
"jti": str(uuid.uuid4()),
"iat": now,
"exp": now + ttl,
"groups": groups or [],
}
return jwt.encode(payload, _hub_sk_pem, algorithm="EdDSA")
def decode_access_token(token: str) -> dict:
"""Verify and decode an access token. Raises on failure."""
if _hub_pk_pem is None:
raise RuntimeError("Hub keypair not loaded")
return jwt.decode(token, _hub_pk_pem, algorithms=["EdDSA"])
# ── Refresh tokens ────────────────────────────────────────────────────────────
def generate_refresh_token() -> tuple[str, str]:
"""Return (raw_token, token_hash). Store hash; give raw to client."""
raw = base64.urlsafe_b64encode(os.urandom(32)).decode()
hashed = blake3.blake3(raw.encode()).hexdigest()
return raw, hashed
def hash_refresh_token(raw: str) -> str:
return blake3.blake3(raw.encode()).hexdigest()
|