path: root/packages/meshbay-cli/src/meshbay_cli/client/keys.py

# SPDX-License-Identifier: LGPL-3.0-or-later
"""
The keys a member derives from a passphrase, and the identities they open.

Byte for byte what `static/keyderive.js` does (docs/MESHBAY_DESIGN.md §3.1,
§3.7); `test_bundle_kdf_parity.py` holds this module to it. A mismatch does not
look like an error — it looks like an account nobody can open.
"""

import base64
import hashlib
import json
import os
from dataclasses import dataclass

from argon2.low_level import Type, hash_secret_raw
from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
from cryptography.hazmat.primitives.asymmetric.x25519 import X25519PrivateKey
from cryptography.hazmat.primitives.ciphers.aead import AESGCM
from cryptography.hazmat.primitives.kdf.hkdf import HKDF
from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2HMAC

PBKDF2_ITERATIONS = 600_000
ARGON2_MEM_KIB = 131072   # 128 MB
ARGON2_TIME = 3
ARGON2_LANES = 1

BUNDLE_MAGIC = b"MBK3"
LEGACY_MAGIC = b"MBK2"    # TRANSITIONAL — read, never written


def b64(data: bytes) -> str:
    return base64.b64encode(data).decode()


def unb64(text: str) -> bytes:
    return base64.b64decode(text)


def hkdf(key: bytes, info: str) -> bytes:
    return HKDF(algorithm=hashes.SHA256(), length=32, salt=None,
                info=info.encode()).derive(key)


def auth_key(username: str, passphrase: str) -> str:
    """What the hub checks at sign-in. The passphrase itself never leaves here."""
    salt = hashlib.sha256(f"meshbay:auth:v1:{username}".encode()).digest()
    return b64(PBKDF2HMAC(algorithm=hashes.SHA512(), length=32, salt=salt,
                          iterations=PBKDF2_ITERATIONS).derive(passphrase.encode()))


def passphrase_half(username: str, passphrase: str) -> bytes:
    """`A`: Argon2id of the passphrase, salted by the account name."""
    salt = hashlib.sha256(f"meshbay:bundle:v2:{username}".encode()).digest()[:16]
    return hash_secret_raw(passphrase.encode(), salt, time_cost=ARGON2_TIME,
                           memory_cost=ARGON2_MEM_KIB, parallelism=ARGON2_LANES,
                           hash_len=32, type=Type.ID)


def bundle_master_key(a: bytes, pepper: bytes, user_id: str) -> bytes:
    """`M`, from `A` and the pepper the hub hands out at sign-in."""
    return hkdf(a + pepper, f"meshbay:bundle-master:v3|{user_id}")


def node_bundle_key(master: bytes, node_pk: str) -> bytes:
    return hkdf(master, f"meshbay:bundle:v3|node|{node_pk}")


class BundleError(Exception):
    pass


def open_bundle(bundle_b64: str, *, master: bytes | None, legacy: bytes | None,
                user_id: str, node_pk: str) -> dict:
    """
    The `{skEd, skX}` sealed in a keypair bundle.

    `MBK3` opens under the node's key derived from `M`, with the account and the
    node as associated data. `MBK2` opens under `A` alone; it is read so a member
    is not locked out of a node still holding one, and never written — an `MBK3`
    replaces it where this machine leaves bundles (`browser-access on`). Anything
    older is refused by name.
    """
    try:
        raw = unb64(bundle_b64)
    except Exception as e:
        raise BundleError("the node's copy of your keys is unreadable") from e
    try:
        if raw[:4] == BUNDLE_MAGIC:
            if master is None:
                raise BundleError("no bundle key in this session")
            aad = f"meshbay:bundle:v3|{user_id}|{node_pk}".encode()
            plain = AESGCM(node_bundle_key(master, node_pk)).decrypt(raw[5:17], raw[17:], aad)
        elif raw[:4] == LEGACY_MAGIC:
            if legacy is None:
                raise BundleError("this node holds an older copy of your keys; sign in "
                                  "again with your passphrase to open it")
            plain = AESGCM(legacy).decrypt(raw[4:16], raw[16:], None)
        else:
            raise BundleError("this node holds your keys in a format no longer read; "
                              "ask its operator to unpin you and send a fresh code")
    except BundleError:
        raise
    except Exception as e:
        raise BundleError("the node's copy of your keys does not open with this "
                          "passphrase") from e
    return json.loads(plain)


def seal_bundle(keys: dict, *, master: bytes, pepper_version: int, user_id: str,
                node_pk: str) -> str:
    """`{skEd, skX}` sealed as `MBK3`, as `keyring.js` `seal` writes it: the magic,
    the pepper version, a 96-bit nonce, then AES-GCM under the node's bundle key
    with the account and the node as associated data."""
    nonce = os.urandom(12)
    aad = f"meshbay:bundle:v3|{user_id}|{node_pk}".encode()
    plain = json.dumps({"skEd": keys["skEd"], "skX": keys["skX"]}).encode()
    ct = AESGCM(node_bundle_key(master, node_pk)).encrypt(nonce, plain, aad)
    return b64(BUNDLE_MAGIC + bytes([pepper_version & 0xFF]) + nonce + ct)


def bundle_is_current(bundle_b64: str, *, master: bytes, user_id: str,
                      node_pk: str) -> bool:
    """Whether a bundle is `MBK3` and opens under this `M`: one sealed under an
    earlier passphrase, or `MBK2`, is to be sealed again."""
    try:
        raw = unb64(bundle_b64)
        if raw[:4] != BUNDLE_MAGIC:
            return False
        aad = f"meshbay:bundle:v3|{user_id}|{node_pk}".encode()
        AESGCM(node_bundle_key(master, node_pk)).decrypt(raw[5:17], raw[17:], aad)
        return True
    except Exception:
        return False


def opens_under(bundle_b64: str, master: bytes, *, user_id: str,
                node_pk: str) -> dict | None:
    """The `{skEd, skX}` of an `MBK3` bundle sealed under `master`, else None."""
    try:
        raw = unb64(bundle_b64)
        if raw[:4] != BUNDLE_MAGIC:
            return None
        aad = f"meshbay:bundle:v3|{user_id}|{node_pk}".encode()
        return json.loads(AESGCM(node_bundle_key(master, node_pk)).decrypt(
            raw[5:17], raw[17:], aad))
    except Exception:
        return None


# ── The recovery key (docs/MESHBAY_DESIGN.md §3.6) ───────────────────────────
#
# 32 random bytes `R`, shown as 13 groups of four Base32 characters; the key
# that seals a recovery copy is HKDF(R, "meshbay:recovery:v1:" + username),
# used as it is — not per node — with the bundle's associated data and pepper
# version 0. As `keyderive.js` `generateRecoveryKey`/`deriveRecoveryKey`.

_B32 = "ABCDEFGHIJKLMNOPQRSTUVWXYZ234567"


def generate_recovery() -> tuple[bytes, str]:
    """`R`, and the grouped mnemonic a person keeps."""
    raw = os.urandom(32)
    return raw, recovery_mnemonic(raw)


def recovery_mnemonic(raw: bytes) -> str:
    text = base64.b32encode(raw).decode().rstrip("=")
    return " ".join(text[i:i + 4] for i in range(0, len(text), 4))


def recovery_bytes(mnemonic: str) -> bytes:
    """`R` from its mnemonic, however spaced or cased; ValueError if it is not one."""
    clean = "".join(c for c in mnemonic.upper() if c.isalnum())
    if any(c not in _B32 for c in clean):
        raise ValueError("a recovery key is letters A to Z and digits 2 to 7")
    raw = base64.b32decode(clean + "=" * (-len(clean) % 8)) if clean else b""
    if len(raw) < 32:
        raise ValueError("this recovery key is too short")
    return raw[:32]


def recovery_key(raw: bytes, username: str) -> bytes:
    return hkdf(raw, f"meshbay:recovery:v1:{username}")


def seal_recovery(keys: dict, *, rkey: bytes, user_id: str, node_pk: str) -> str:
    nonce = os.urandom(12)
    aad = f"meshbay:bundle:v3|{user_id}|{node_pk}".encode()
    plain = json.dumps({"skEd": keys["skEd"], "skX": keys["skX"]}).encode()
    return b64(BUNDLE_MAGIC + bytes([0]) + nonce + AESGCM(rkey).encrypt(nonce, plain, aad))


def open_recovery(bundle_b64: str, *, rkey: bytes, user_id: str, node_pk: str) -> dict:
    try:
        raw = unb64(bundle_b64)
        if raw[:4] != BUNDLE_MAGIC:
            raise ValueError
        aad = f"meshbay:bundle:v3|{user_id}|{node_pk}".encode()
        return json.loads(AESGCM(rkey).decrypt(raw[5:17], raw[17:], aad))
    except Exception as e:
        raise BundleError("the node's recovery copy of your keys does not open with "
                          "this recovery key") from e


@dataclass
class Identity:
    """One account's keys on one node."""

    sk_ed: Ed25519PrivateKey
    sk_x: X25519PrivateKey

    @classmethod
    def from_pkcs8(cls, keys: dict) -> "Identity":
        sk_ed = serialization.load_der_private_key(unb64(keys["skEd"]), password=None)
        sk_x = serialization.load_der_private_key(unb64(keys["skX"]), password=None)
        if not isinstance(sk_ed, Ed25519PrivateKey) or not isinstance(sk_x, X25519PrivateKey):
            raise BundleError("the node's copy of your keys holds the wrong kind of key")
        return cls(sk_ed, sk_x)

    def to_pkcs8(self) -> dict:
        def der(k) -> str:
            return b64(k.private_bytes(serialization.Encoding.DER,
                                       serialization.PrivateFormat.PKCS8,
                                       serialization.NoEncryption()))
        return {"skEd": der(self.sk_ed), "skX": der(self.sk_x)}

    @property
    def pk_ed(self) -> str:
        return b64(self.sk_ed.public_key().public_bytes_raw())

    @property
    def pk_x(self) -> str:
        return b64(self.sk_x.public_key().public_bytes_raw())