path: root/packages/meshbay-cli/src/meshbay_cli/commands/account.py

"""
`account`: the account itself — see it, create one, verify its address, change
or reset the passphrase, change the address, end every session, erase it.

What the application's sign-up form and Profile page do, with the same rules
(docs/MESHBAY_DESIGN.md §3.1, §3.6). The passphrase never leaves this machine:
the hub is shown the key derived from it, and the floor a new one must clear
is checked here, as the page checks it. A new passphrase carries the account's
keys on every node across first, and only then is the hub told — so a change
that cannot be carried leaves the account as it was.
"""

import asyncio
import getpass
import json
import os
import sys
import time
from pathlib import Path

from ..client.account import Account, NoPassphrase
from ..client.groups import my_groups
from ..client.hub import HubError
from ..client.keys import (
    auth_key,
    b64,
    bundle_master_key,
    generate_recovery,
    passphrase_half,
    recovery_bytes,
    recovery_key,
    unb64,
)
from ..client.passphrase import passphrase_problem, solve_captcha
from ..client.rewrap import Carried, carry
from ..client.store import FILE_KEY_NOTICE, KEY_FILE
from ..errors import AUTH, NOT_FOUND, REFUSED, USAGE, CliError
from ..output import confirm, err, safe, sentence, terminal
from .context import Context

HELP = ("your account: show it, register, verify, change or reset the passphrase, "
        "change the e-mail address, sign out everywhere, delete it")
VERBS = ("show", "register", "verify", "passphrase", "reset", "email",
         "sign-out-everywhere", "delete")
USERNAME_MIN = 8          # the hub's USERNAME_MIN_LEN
# How many earlier `M` this machine keeps, to carry copies on nodes that were
# out of reach when the passphrase changed.
KEEP_MASTERS = 5


def add_parser(sub, parent) -> None:
    p = sub.add_parser("account", parents=[parent], help=HELP, description=sentence(HELP)
                       + ". With no verb, shows the account.")
    p.add_argument("verb", nargs="?", default="show", choices=VERBS,
                   help="what to do; show when left out")
    p.add_argument("arg", nargs="?", metavar="CODE|ADDRESS",
                   help="verify: the code e-mailed at registration; email: the new "
                        "address")
    p.add_argument("--email", metavar="ADDRESS",
                   help="register, verify, reset: the account's e-mail address")
    p.add_argument("--code", help="reset: the code e-mailed for it; email: the code sent "
                                  "to the new address")
    p.add_argument("--new-password-file", metavar="FILE",
                   help="passphrase: read the new passphrase from FILE's first line "
                        "(- for stdin); also MESHBAY_NEW_PASSWORD")
    p.add_argument("--recovery-key-file", metavar="FILE",
                   help="reset: read the recovery key from FILE, to restore your keys "
                        "on the nodes")
    p.add_argument("--no-remember", action="store_true",
                   help="reset: sign in afterwards without storing the session")
    p.add_argument("-y", "--yes", action="store_true",
                   help="delete without asking for confirmation")


async def run(ctx: Context) -> int:
    a = ctx.args
    await ctx.hub.check_version()
    verb = a.verb
    if verb == "register":
        return await _register(ctx)
    if verb == "verify":
        return await _verify(ctx)
    if verb == "reset":
        return await _reset(ctx)
    account = ctx.account()
    try:
        if verb == "show":
            return await _show(ctx, account)
        if verb == "passphrase":
            return await _change_passphrase(ctx, account)
        if verb == "email":
            return await _email(ctx, account)
        if verb == "sign-out-everywhere":
            return await _sign_out_everywhere(ctx, account)
        return await _delete(ctx, account)
    finally:
        await account.close()


# ── Reading what a person types ──────────────────────────────────────────────

def _ask(prompt: str, given: str | None, what: str) -> str:
    if given:
        return given.strip()
    if not terminal(sys.stdin):
        raise CliError(USAGE, f"Name the {what}: there is no terminal to ask on.")
    value = input(prompt).strip()
    if not value:
        raise CliError(USAGE, f"No {what} given.")
    return value


def _read_first_line(path: str) -> str:
    try:
        text = sys.stdin.read() if path == "-" else Path(path).read_text(encoding="utf-8")
    except OSError as e:
        raise CliError(USAGE, f"Cannot read {path}: {e.strerror}") from e
    return text.splitlines()[0] if text else ""


def _new_passphrase(given: str | None) -> str:
    """A new passphrase: `given`, else typed twice at a terminal; held to the floor."""
    if given is None:
        if not terminal(sys.stdin):
            raise NoPassphrase("A new passphrase is needed and there is no terminal to "
                               "ask on.")
        given = getpass.getpass("New passphrase: ")
        if problem := passphrase_problem(given):
            raise CliError(USAGE, problem)
        if getpass.getpass("The same again: ") != given:
            raise CliError(USAGE, "The two passphrases differ; nothing was changed.")
    if problem := passphrase_problem(given):
        raise CliError(USAGE, problem)
    return given


def _given_passphrase(ctx: Context) -> str | None:
    """The global sources of a passphrase, without prompting."""
    a = ctx.args
    if a.password is not None:
        return a.password
    if a.password_file:
        return _read_first_line(a.password_file)
    return os.environ.get("MESHBAY_PASSWORD")


async def _captcha(ctx: Context) -> str | None:
    challenge = await ctx.hub.captcha()
    if challenge is None:
        return None
    token = await asyncio.to_thread(solve_captcha, challenge)
    if token is None:
        raise CliError(REFUSED, "The hub's proof of work has no answer; try again.")
    return token


def _hub_refusal(e: HubError, what: str) -> CliError:
    if e.status == 429:
        return CliError(REFUSED, f"{what}: too many attempts — {e.detail}")
    if e.status in (401, 403):
        return CliError(AUTH, f"{what}: {e.detail}")
    if e.status == 404:
        return CliError(NOT_FOUND, f"{what}: {e.detail}")
    return CliError(REFUSED, f"{what}: {e.detail}")


def _say(ctx: Context, data: dict, text: str) -> None:
    if ctx.args.json:
        print(json.dumps(data, ensure_ascii=False))
    elif not ctx.args.quiet:
        print(text)


# ── show ─────────────────────────────────────────────────────────────────────

async def _show(ctx: Context, account: Account) -> int:
    me = await account.call(account.hub.me)
    if ctx.args.json:
        print(json.dumps({"hub": ctx.hub.url, **{k: me.get(k) for k in (
            "user_id", "username", "email", "role", "status")}}, ensure_ascii=False))
        return 0
    print(f"Account:  {safe(me.get('username', ''))}\n"
          f"Hub:      {ctx.hub.url}\n"
          f"E-mail:   {safe(me.get('email', ''))}\n"
          f"Status:   {safe(me.get('status', ''))}")
    if me.get("role") and me["role"] != "user":
        print(f"Role:     {safe(me['role'])}")
    return 0


# ── register, verify ─────────────────────────────────────────────────────────

async def _register(ctx: Context) -> int:
    a = ctx.args
    username = _ask("Username: ", ctx.username, "account name (--user)")
    if len(username) < USERNAME_MIN or len(username) > 64 or \
            not username.replace("_", "").replace("-", "").replace(".", "").isalnum():
        raise CliError(USAGE, f"A username is {USERNAME_MIN} to 64 letters, digits, "
                              "-, _ or .")
    email = _ask("E-mail address: ", a.email, "e-mail address (--email)")
    passphrase = _new_passphrase(_given_passphrase(ctx))
    ak = await asyncio.to_thread(auth_key, username, passphrase)
    raw, mnemonic = generate_recovery()
    try:
        try:
            reg = await ctx.hub.register(username, email, ak, await _captcha(ctx))
        except HubError as e:
            if e.detail not in ("captcha_failed", "captcha_required"):
                raise
            # An answer is spent once checked; a fresh one, once.
            reg = await ctx.hub.register(username, email, ak, await _captcha(ctx))
    except HubError as e:
        raise _hub_refusal(e, "Not registered") from e

    if ctx.args.json:
        print(json.dumps({"user_id": reg.get("user_id"), "username": username,
                          "recovery_key": mnemonic,
                          "email_verification_required": True}))
    else:
        print(f"Account {safe(username)} created on {ctx.hub.url}.\n\n"
              "Your recovery key — shown this once, kept nowhere by MeshBay, never "
              "sent to the hub:\n\n"
              f"    {mnemonic}\n\n"
              "Keep it somewhere safe, apart from your passphrase. If you forget the "
              "passphrase, it is what gives you your groups back.")
    del raw
    if not terminal(sys.stdin) or ctx.args.json:
        err(f"A code was sent to {safe(email)}. Then: meshbay-cli account verify CODE "
            f"--email {safe(email)}")
        return 0
    code = input(f"\nThe code sent to {email} (Enter to do it later): ").strip()
    if not code:
        err(f"Later: meshbay-cli account verify CODE --email {safe(email)}")
        return 0
    await _verify_code(ctx, email, code)
    return 0


async def _verify(ctx: Context) -> int:
    a = ctx.args
    code = _ask("The code: ", a.arg or a.code, "code")
    email = _ask("E-mail address: ", a.email, "e-mail address (--email)")
    await _verify_code(ctx, email, code)
    return 0


async def _verify_code(ctx: Context, email: str, code: str) -> None:
    try:
        await ctx.hub.verify_email(email, code)
    except HubError as e:
        raise _hub_refusal(e, "Not verified") from e
    _say(ctx, {"verified": True, "email": email},
         "Verified. Sign in with: meshbay-cli login")


# ── passphrase: change a known one ───────────────────────────────────────────

async def _change_passphrase(ctx: Context, account: Account) -> int:
    a = ctx.args
    old = ctx.passphrase()
    given = (_read_first_line(a.new_password_file) if a.new_password_file
             else os.environ.get("MESHBAY_NEW_PASSWORD"))
    new = _new_passphrase(given)
    if new == old:
        raise CliError(USAGE, "The new passphrase is the current one; nothing was changed.")
    me = await account.call(account.hub.me)
    if int(me.get("passphrase_locked_for") or 0) > 0:
        raise CliError(AUTH, "Sign-in is locked for this account for "
                             f"{max(1, int(me['passphrase_locked_for']) // 60)} min; "
                             "nothing was changed.")
    user, uid = account.username, me["user_id"]
    old_ak = await asyncio.to_thread(auth_key, user, old)
    try:
        pep = await account.call(account.hub.bundle_pepper, old_ak)
    except HubError as e:
        if e.status == 403:
            raise CliError(AUTH, "The current passphrase does not match; nothing was "
                                 "changed.") from e
        raise _hub_refusal(e, "Not changed") from e
    pepper, version = unb64(pep["bundle_pepper"]), int(pep.get("bundle_pepper_version") or 1)
    old_a = await asyncio.to_thread(passphrase_half, user, old)
    new_a = await asyncio.to_thread(passphrase_half, user, new)
    old_m, new_m = (bundle_master_key(old_a, pepper, uid),
                    bundle_master_key(new_a, pepper, uid))
    # What opens the copies as they are: the current passphrase's keys.
    await account.update(lambda st, se: se.update(
        master=b64(old_m), legacy=b64(old_a), pepper_version=version))

    groups = await my_groups(account)
    progress = None if a.quiet or a.json else (lambda g: err(f"  {safe(g)}..."))
    if progress:
        err("Sealing your keys on each node under the new passphrase:")
    carried = await carry(account, groups, master=new_m, pepper_version=version,
                          progress=progress)
    new_ak = await asyncio.to_thread(auth_key, user, new)
    try:
        r = await account.call(account.hub.change_passphrase, old_ak, new_ak)
    except (HubError, CliError) as e:
        # The hub kept the old passphrase: the copies go back under it, opened
        # with the new one they are sealed under now.
        await account.update(lambda st, se: se.update(master=b64(new_m), legacy=b64(new_a)))
        try:
            back = await carry(account, groups, master=old_m, pepper_version=version,
                               only=set(carried.done))
        finally:
            await account.update(lambda st, se: se.update(master=b64(old_m),
                                                          legacy=b64(old_a)))
        detail = getattr(e, "detail", None) or getattr(e, "message", str(e))
        lost = len(carried.done) - len(back.done)
        raise CliError(REFUSED, f"The hub refused the new passphrase ({detail}); "
                                "nothing was changed" + (
                                    f", but {lost} node(s) could not be set back and "
                                    "hold your keys under the new one" if lost else "")
                       + ".") from e

    def commit(state: dict, secrets: dict) -> None:
        kept = [m for m in secrets.get("previous_masters", []) if m != b64(old_m)]
        secrets.update(master=b64(new_m), legacy=b64(new_a), pepper_version=version,
                       previous_masters=([b64(old_m)] + kept)[:KEEP_MASTERS],
                       access_token=r["access_token"], refresh_token=r["refresh_token"])
        secrets["access_exp"] = time.time() + int(r.get("expires_in", 0))
    await account.update(commit)
    _report(ctx, carried, "Passphrase changed", changed=True)
    return 0


def _report(ctx: Context, carried: Carried, head: str, *, changed: bool) -> None:
    if ctx.args.json:
        print(json.dumps({"changed": changed, "nodes_sealed": len(carried.done),
                          "unreachable": carried.unreachable, "failed": carried.failed}))
        return
    print(f"{head}. Your keys were sealed again on {len(carried.done)} node"
          f"{'s' if len(carried.done) != 1 else ''}.")
    if carried.stragglers:
        err("Not reached — " + ("this machine carries them over the next time it "
                                "connects there:" if changed else
                                "these groups may need their operator (`member unpin`, "
                                "then a fresh code):"))
        for s in carried.stragglers:
            err(f"  {safe(s)}")


# ── reset: a forgotten passphrase ────────────────────────────────────────────

async def _reset(ctx: Context) -> int:
    a = ctx.args
    username = _ask("Username: ", ctx.username or ctx.stored_username,
                    "account name (--user)")
    code = a.code
    if not code:
        email = _ask("The account's e-mail address: ", a.email, "e-mail address (--email)")
        try:
            await ctx.hub.reset_request(username, email, await _captcha(ctx))
        except HubError as e:
            raise _hub_refusal(e, "No code sent") from e
        err(f"If {safe(username)} and {safe(email)} are an account's, a code is on its "
            "way there (valid one hour).")
        if not terminal(sys.stdin):
            err(f"Then: meshbay-cli account reset -u {safe(username)} --code CODE")
            return 0
        code = _ask("The code: ", None, "code")
    passphrase = _new_passphrase(_given_passphrase(ctx))
    recovery = None
    if a.recovery_key_file:
        text = (sys.stdin.read() if a.recovery_key_file == "-"
                else Path(a.recovery_key_file).read_text(encoding="utf-8"))
        recovery = text.strip()
    elif terminal(sys.stdin):
        recovery = input("Your recovery key (Enter if you have none): ").strip()
    rkey = None
    if recovery:
        try:
            rkey = recovery_key(recovery_bytes(recovery), username)
        except ValueError as e:
            raise CliError(USAGE, f"{sentence(str(e))}; nothing was changed.") from e

    new_ak = await asyncio.to_thread(auth_key, username, passphrase)
    try:
        await ctx.hub.reset_passphrase(username, code.strip(), new_ak)
    except HubError as e:
        raise _hub_refusal(e, "Not reset") from e

    # Signed in afresh, as `login` does: the reset ended every session and
    # dropped every device key on the hub.
    remember = not a.no_remember
    account = Account(ctx.hub, username, lambda: passphrase)
    stored = ctx.store.peek()
    if stored.get("hub") == ctx.hub.url and stored.get("username") == username:
        try:
            ctx.store.load()
            account.state, account.secrets = ctx.store.state, ctx.store.secrets
        except CliError:
            pass
    try:
        await account.sign_in(remember=remember)
        if remember:
            if await account.remember(ctx.store) == KEY_FILE:
                err(FILE_KEY_NOTICE)
        progress = None if a.quiet or a.json else (lambda g: err(f"  {safe(g)}..."))
        if progress:
            err("Restoring your keys on each node:")
        carried = await carry(account, await my_groups(account), master=account.master,
                              pepper_version=account.pepper_version, recovery=rkey,
                              own_keys=account.browser_access == "on", progress=progress)
    finally:
        await account.close()
    _report(ctx, carried, "Passphrase reset, and signed in", changed=False)
    if not rkey and not ctx.args.json:
        err("Without the recovery key, groups whose keys this machine does not keep "
            "need their operator (`member unpin`, then a fresh code).")
    return 0


# ── email, sign-out-everywhere, delete ───────────────────────────────────────

async def _email(ctx: Context, account: Account) -> int:
    a = ctx.args
    if a.code:
        try:
            r = await account.call(account.hub.verify_email_change, a.code.strip())
        except HubError as e:
            raise _hub_refusal(e, "Not changed") from e
        _say(ctx, {"email": r.get("email")}, f"E-mail address changed to "
                                              f"{safe(r.get('email', ''))}.")
        return 0
    address = _ask("New e-mail address: ", a.arg, "new address")
    ak = await asyncio.to_thread(auth_key, account.username, ctx.passphrase())
    try:
        await account.call(account.hub.change_email, address, ak)
    except HubError as e:
        raise _hub_refusal(e, "Not changed") from e
    if ctx.args.json or not terminal(sys.stdin):
        _say(ctx, {"email_verification_required": True, "email": address},
             f"A code was sent to {safe(address)}. Then: meshbay-cli account email "
             "--code CODE")
        return 0
    code = input(f"The code sent to {address}: ").strip()
    if not code:
        err("Later: meshbay-cli account email --code CODE")
        return 0
    try:
        r = await account.call(account.hub.verify_email_change, code)
    except HubError as e:
        raise _hub_refusal(e, "Not changed") from e
    print(f"E-mail address changed to {safe(r.get('email', ''))}.")
    return 0


async def _sign_out_everywhere(ctx: Context, account: Account) -> int:
    r = await account.call(account.hub.revoke_sessions)
    n = int(r.get("revoked") or 0)
    _say(ctx, {"revoked": n},
         f"Signed out everywhere: {n} session{'s' if n != 1 else ''} ended. Devices "
         "with a key of their own (this machine, if it is remembered) sign back in with "
         "it; remove one from your devices to stop it.")
    return 0


async def _delete(ctx: Context, account: Account) -> int:
    a = ctx.args
    confirm(f"Erase the account {account.username} on {ctx.hub.url}, and every group "
            "it owns? This cannot be undone.", a.yes)
    ak = await asyncio.to_thread(auth_key, account.username, ctx.passphrase())
    try:
        r = await account.call(account.hub.delete_account, ak)
    except HubError as e:
        if e.status == 403:
            raise CliError(AUTH, "The passphrase does not match; nothing was erased.") from e
        raise _hub_refusal(e, "Not erased") from e
    if account.remembered:
        with ctx.store.locked():
            ctx.store.erase()
    gone = r.get("groups_deleted") or []
    _say(ctx, {"deleted": True, "groups_deleted": gone},
         f"Account {safe(account.username)} erased"
         + (f", with {len(gone)} group{'s' if len(gone) != 1 else ''} it owned"
            if gone else "") + ".")
    return 0