""" `devices`: this account's devices on the node serving a group — what the application's "Your devices on this node" shows and does. Identities are per node, so a group names the node. `approve` admits the device showing a code; `revoke` retires one, asked first (`--yes` for a script). A node where this machine is the only linked device admits no other without this. """ import json import sys from ..client import devices from ..client.groups import my_groups, open_group from ..client.paths import group_arg, handle, resolve_group from ..errors import REFUSED, USAGE, CliError from ..output import confirm, err, safe, sentence from .context import Context HELP = "your devices on the node serving a group: list, approve a new one, revoke one" def add_parser(sub, parent) -> None: p = sub.add_parser("devices", parents=[parent], help=HELP, description=sentence(HELP) + ". A new device shows a code; approve it here, from a machine " "already linked to that node.") p.add_argument("verb", choices=("list", "approve", "revoke")) p.add_argument("group", metavar="GROUP", help="the group, as `groups` lists it") p.add_argument("arg", nargs="?", metavar="CODE|DEVICE", help="approve: the code the new device shows; revoke: the device, by " "the first characters of its key as `devices list` shows it") p.add_argument("-y", "--yes", action="store_true", help="revoke without asking for confirmation") def _when(iso: str) -> str: return (iso or "")[:16].replace("T", " ") async def run(ctx: Context) -> int: a = ctx.args if a.verb != "list" and not a.arg: raise CliError(USAGE, f"Name the {'code' if a.verb == 'approve' else 'device'}: " f"meshbay-cli devices {a.verb} GROUP " f"{'CODE' if a.verb == 'approve' else 'DEVICE'}") await ctx.hub.check_version() account = ctx.account() session = None try: group = resolve_group(await my_groups(account), group_arg(a.group)) name = handle(group) session = await open_group(account, group) if a.verb == "list": return _print(ctx, name, await devices.list_devices(session)) if a.verb == "approve": pk = await devices.approve(session, account, a.arg) print(json.dumps({"approved": pk}) if a.json else f"Approved the device {pk[:16]} on the node serving {safe(name)}.") return 0 device = devices.find_device((await devices.list_devices(session))["devices"], a.arg) if device.get("is_this_one"): raise CliError(REFUSED, "That is this machine. Revoke it from another of your " "devices.") label = device.get("label") or "no label" confirm(f"Revoke the device {device['pk_ed25519'][:16]} ({safe(label)}) on the " f"node serving {safe(name)}? It will no longer open this node's groups.", a.yes) await devices.revoke(session, account, device["pk_ed25519"]) print(json.dumps({"revoked": device["pk_ed25519"]}) if a.json else f"Revoked the device {device['pk_ed25519'][:16]}.") return 0 finally: if session is not None: await session.close() await account.close() def _print(ctx: Context, name: str, listed: dict) -> int: rows = listed["devices"] if ctx.args.json: json.dump(listed, sys.stdout, indent=1) print() return 0 print(f" {'KEY':<18}{'ADDED':<18}{'VIA':<12}LABEL") for d in rows: mark = "*" if d.get("is_this_one") else " " print(f"{mark} {d['pk_ed25519'][:16]:<18}{_when(d.get('pinned_at')):<18}" f"{safe(d.get('pinned_via') or ''):<12}{safe(d.get('label') or '')}") if not ctx.args.quiet: err("* this machine") if listed["pending"]: n = listed["pending"] err(f"{n} device{'s' if n != 1 else ''} waiting for approval: " f"meshbay-cli devices approve {name} CODE") return 0