""" 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 asyncio import base64 import hashlib import os import time import uuid from concurrent.futures import ThreadPoolExecutor 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.ciphers.aead import AESGCM from cryptography.hazmat.primitives.kdf.argon2 import Argon2id from cryptography.hazmat.primitives.kdf.hkdf import HKDF from cryptography.hazmat.primitives.hashes import SHA256 # Argon2id parameters — versioned for gradual migration _ARGON2_LANES = 4 _ARGON2_KEY_LEN = 32 _ARGON2_VERSIONS = { 1: {"iterations": 3, "memory_cost": 65536}, # 64 MB — initial 2: {"iterations": 3, "memory_cost": 262144}, # 256 MB — raw password (legacy) 3: {"iterations": 3, "memory_cost": 262144}, # 256 MB — auth_key input (password split) # 64 MiB, t=3, p=4 — RFC 9106's second recommended setting. What this hashes # is already PBKDF2-SHA512 at 600 000 iterations of the passphrase, done by # the client, and online guessing is bounded by the sign-in lockout, so the # memory above this bought a constant factor against an offline attacker # with the database, at four times the cost of every sign-in. Versions 3 and # above are the auth_key scheme; a v3 hash is rewritten at its next sign-in. 4: {"iterations": 3, "memory_cost": 65536}, } _ARGON2_CURRENT_VERSION = 4 # 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" _email_key: bytes | None = None # ── 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, _email_key _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 sk_raw = sk.private_bytes( serialization.Encoding.Raw, serialization.PrivateFormat.Raw, serialization.NoEncryption(), ) _email_key = HKDF( algorithm=SHA256(), length=32, salt=None, info=b"meshbay:email:v1", ).derive(sk_raw) 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 def hub_private_key_pem() -> bytes: if _hub_sk_pem is None: raise RuntimeError("Hub keypair not loaded — call load_hub_keypair() first") return _hub_sk_pem def hub_id() -> str: """This hub's configured identity. An accessor, not the module global, because `load_hub_keypair` runs at startup and every one of these is set *after* import. A module that wrote `from meshbay_hub.auth import _hub_id` captured the default and kept it: `federation.py` did, so it signed with a `None` key and announced itself as `meshbay.org` whatever its configuration said. Reading through a function is what makes "call once at startup" true for readers as well as for the writer. """ return _hub_id # ── Password ────────────────────────────────────────────────────────────────── def hash_password(password: str, version: int = _ARGON2_CURRENT_VERSION) -> tuple[bytes, bytes]: """Hash with the parameters of `version`, which is what the caller stores. The version is an argument because the stored `pw_version` is what verification reads the parameters from: hashing at one version's parameters and recording another makes an account nobody can sign in to. That mistake sat unseen while versions 2 and 3 shared their parameters. """ salt = os.urandom(16) params = _ARGON2_VERSIONS[version] pw_hash = Argon2id( salt=salt, length=_ARGON2_KEY_LEN, iterations=params["iterations"], lanes=_ARGON2_LANES, memory_cost=params["memory_cost"], ).derive(password.encode()) return pw_hash, salt def verify_password(password: str, pw_hash: bytes, salt: bytes, version: int = 2) -> bool: params = _ARGON2_VERSIONS.get(version, _ARGON2_VERSIONS[_ARGON2_CURRENT_VERSION]) try: Argon2id( salt=salt, length=_ARGON2_KEY_LEN, iterations=params["iterations"], lanes=_ARGON2_LANES, memory_cost=params["memory_cost"], ).verify(password.encode(), pw_hash) return True except Exception: return False # ── Argon2 off the event loop, one at a time ───────────────────────────────── # # One derivation is 64 MiB and ~0.1 s on meshbay.org (0.45 s for an old 256 MB # hash, until its account next signs in). Called from an # async handler it stops the whole hub for that long — every request, every node # socket, every offer relayed — once per sign-in, passphrase change, reset and # registration. # # It cannot simply go to a thread pool: **two concurrent derivations with # `lanes=4` deadlock inside OpenSSL** and never return, at no CPU (measured # 2026-09-14 on cryptography 50.0.x / OpenSSL 4.0.x, here and on meshbay.org; # `lanes=1` does not, and `lanes` is part of every stored hash, so it is not # ours to change). Inline on the loop they could never overlap, which is the only # reason this never hung in production. # # So: a dedicated executor with exactly one worker. The loop is free while a # derivation runs, and derivations still never overlap — including when the # request that queued one is cancelled mid-way, which a semaphore around # `to_thread` would get wrong (the permit is released while the thread is still # deriving, and the next one starts beside it). It also bounds Argon2's memory # to one derivation, whatever the number of callers. _argon2_executor = ThreadPoolExecutor(max_workers=1, thread_name_prefix="argon2") async def hash_password_off_loop(password: str, version: int = _ARGON2_CURRENT_VERSION) -> tuple[bytes, bytes]: loop = asyncio.get_running_loop() return await loop.run_in_executor(_argon2_executor, hash_password, password, version) async def verify_password_off_loop(password: str, pw_hash: bytes, salt: bytes, version: int = 2) -> bool: loop = asyncio.get_running_loop() return await loop.run_in_executor( _argon2_executor, verify_password, password, pw_hash, salt, version) def pw_needs_rehash(version: int) -> bool: return version < _ARGON2_CURRENT_VERSION def current_pw_version() -> int: return _ARGON2_CURRENT_VERSION # ── JWT ─────────────────────────────────────────────────────────────────────── def issue_access_token( user_id: str, ttl: int = 3600, groups: list[str] | None = None, scope: str = "user", ) -> str: """ Issue a signed JWT access token. Carries no user key. It used to, and the node recorded that key as the uploader's identity — so the party issuing tokens decided who could delete a file. The hub certifies accounts; nodes pin keys. 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). scope: "user" (browser, full access) or "node" (daemon, restricted). """ if _hub_sk_pem is None: raise RuntimeError("Hub keypair not loaded") now = int(time.time()) payload = { "iss": _hub_id, "sub": user_id, "hub_id": _hub_id, "jti": str(uuid.uuid4()), "iat": now, "exp": now + ttl, "groups": groups or [], "scope": scope, } 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") # Clock-skew tolerance (meshbay_common.handshake.JWT_LEEWAY_SECONDS): a # client whose clock is a little fast must still be able to call the API. return jwt.decode(token, _hub_pk_pem, algorithms=["EdDSA"], leeway=60) # ── Email encryption at rest ────────────────────────────────────────────────── def encrypt_email(plaintext: str) -> str: """Encrypt an email address for storage. Returns base64(nonce + ciphertext).""" if _email_key is None: raise RuntimeError("Hub keypair not loaded") nonce = os.urandom(12) ct = AESGCM(_email_key).encrypt(nonce, plaintext.encode(), None) return base64.b64encode(nonce + ct).decode() def decrypt_email(stored: str) -> str: """Decrypt an email address from storage.""" if _email_key is None: raise RuntimeError("Hub keypair not loaded") raw = base64.b64decode(stored) nonce, ct = raw[:12], raw[12:] return AESGCM(_email_key).decrypt(nonce, ct, None).decode() def hash_email_blind(email: str) -> str: """Deterministic HMAC-SHA256 of the lowercased email for uniqueness checks. The encrypted email uses a random nonce, so two encryptions of the same address produce different ciphertexts. This blind index allows a DB-level uniqueness constraint without decrypting every row. """ if _email_key is None: raise RuntimeError("Hub keypair not loaded") import hmac as _hmac return _hmac.new( _email_key, email.strip().lower().encode(), hashlib.sha256, ).hexdigest() # ── 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()