path: root/packages/meshbay-hub/src/meshbay_hub/auth.py

"""
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 import serialization
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
from cryptography.hazmat.primitives.ciphers.aead import AESGCM
from cryptography.hazmat.primitives.hashes import SHA256
from cryptography.hazmat.primitives.kdf.argon2 import Argon2id
from cryptography.hazmat.primitives.kdf.hkdf import HKDF

# 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()