""" 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.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 — production target 3: {"iterations": 3, "memory_cost": 262144}, # 256 MB — auth_key input (password split) } _ARGON2_CURRENT_VERSION = 3 # 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 # ── Password ────────────────────────────────────────────────────────────────── def hash_password(password: str) -> tuple[bytes, bytes]: """Hash a password with Argon2id (current version). Returns (hash, salt).""" salt = os.urandom(16) params = _ARGON2_VERSIONS[_ARGON2_CURRENT_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 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") return jwt.decode(token, _hub_pk_pem, algorithms=["EdDSA"]) # ── 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()